feat(worker-datasets-raw): add verification option for raw materialize jobs

crates/bin/ampctl/src/cmd/dataset/deploy.rs

@@ -64,6 +64,16 @@ pub struct Args {
     /// The worker must be active (has sent heartbeats recently) for the deployment to succeed.
     #[arg(long, value_parser = clap::value_parser!(NodeSelector))]
     pub worker_id: Option<NodeSelector>,
+
+    /// Enable cryptographic verification of EVM block data during extraction
+    ///
+    /// When enabled, verifies block hashes, transaction roots, and receipt roots
+    /// before writing data to storage. Only applicable to EVM raw datasets.
+    /// Verification failures are retryable errors.
+    ///
+    /// Defaults to false if not specified.
+    #[arg(long)]
+    pub verify: bool,
 }
 
 /// Deploy a dataset to start syncing blockchain data.
@@ -81,17 +91,27 @@ pub async fn run(
         end_block,
         parallelism,
         worker_id,
+        verify,
     }: Args,
 ) -> Result<(), Error> {
     tracing::debug!(
         %dataset_ref,
         ?end_block,
         %parallelism,
         ?worker_id,
+        %verify,
         "Deploying dataset"
     );
 
-    let job_id = deploy_dataset(&global, &dataset_ref, end_block, parallelism, worker_id).await?;
+    let job_id = deploy_dataset(
+        &global,
+        &dataset_ref,
+        end_block,
+        parallelism,
+        worker_id,
+        verify,
+    )
+    .await?;
     let result = DeployResult { job_id };
     global.print(&result).map_err(Error::JsonSerialization)?;
 
@@ -102,18 +122,19 @@ pub async fn run(
 ///
 /// POSTs to the versioned `/datasets/{namespace}/{name}/versions/{version}/deploy` endpoint
 /// and returns the job ID.
-#[tracing::instrument(skip_all, fields(%dataset_ref, ?end_block, %parallelism, ?worker_id))]
+#[tracing::instrument(skip_all, fields(%dataset_ref, ?end_block, %parallelism, ?worker_id, %verify))]
 async fn deploy_dataset(
     global: &GlobalArgs,
     dataset_ref: &Reference,
     end_block: Option<EndBlock>,
     parallelism: u16,
     worker_id: Option<NodeSelector>,
+    verify: bool,
 ) -> Result<JobId, Error> {
     let client = global.build_client().map_err(Error::ClientBuild)?;
     let job_id = client
         .datasets()
-        .deploy(dataset_ref, end_block, parallelism, worker_id)
+        .deploy(dataset_ref, end_block, parallelism, worker_id, verify)
         .await
         .map_err(Error::Deploy)?;
 

crates/bin/ampctl/src/cmd/dataset/deploy__after_help.md

@@ -20,6 +20,11 @@ Deploy dataset, staying 100 blocks behind chain tip:
 ampctl dataset deploy my_namespace/my_dataset@1.0.0 --end-block -100
 ```
 
+Deploy EVM dataset with cryptographic verification:
+```
+ampctl dataset deploy my_namespace/my_dataset@1.0.0 --verify
+```
+
 Use custom admin URL:
 ```
 ampctl dataset deploy my_namespace/my_dataset@1.0.0 --admin-url http://production:1610

crates/clients/admin/src/datasets.rs

@@ -268,6 +268,7 @@ impl<'a> DatasetsClient<'a> {
         end_block: Option<EndBlock>,
         parallelism: u16,
         worker_id: Option<NodeSelector>,
+        verify: bool,
     ) -> Result<JobId, DeployError> {
         let namespace = dataset_ref.namespace();
         let name = dataset_ref.name();
@@ -285,6 +286,7 @@ impl<'a> DatasetsClient<'a> {
             end_block,
             parallelism,
             worker_id,
+            verify,
         };
 
         let response = self
@@ -1283,6 +1285,8 @@ struct DeployRequest {
     parallelism: u16,
     #[serde(skip_serializing_if = "Option::is_none")]
     worker_id: Option<NodeSelector>,
+    #[serde(skip_serializing_if = "std::ops::Not::not")]
+    verify: bool,
 }
 
 /// Input type for dataset registration manifest parameter.

crates/core/datasets-raw/src/rows.rs

@@ -35,6 +35,15 @@ impl IntoIterator for Rows {
     }
 }
 
+impl<'a> IntoIterator for &'a Rows {
+    type Item = &'a TableRows;
+    type IntoIter = std::slice::Iter<'a, TableRows>;
+
+    fn into_iter(self) -> Self::IntoIter {
+        self.0.iter()
+    }
+}
+
 /// A record batch associated with a single block of chain data, for populating raw datasets.
 pub struct TableRows {
     pub table: Table,

crates/core/verification/Cargo.toml

@@ -8,13 +8,15 @@ license-file.workspace = true
 alloy.workspace = true
 amp-client = { path = "../../clients/flight" }
 anyhow.workspace = true
+arrow.workspace = true
 arrow-flight.workspace = true
 clap.workspace = true
 futures.workspace = true
 indicatif.workspace = true
 reqwest.workspace = true
 serde.workspace = true
 serde_json.workspace = true
+thiserror.workspace = true
 tokio.workspace = true
 tonic.workspace = true
 typed-arrow.workspace = true