Celestia: re-add block verification

CHANGELOG.md

@@ -1,3 +1,7 @@
+# 2026-04-20
+- #2764 Celestia adapter: re-adds block integrity verification on fetch. Replaces the previously-reverted boolean toggle (PR #2489 / reverted in PR #2520) with a
+  `verify_on_fetch_mode` enum accepting `"off"` (default), `"log_error"`, or `"return_error"`. `log_error` runs verification and logs via `tracing::error!` on failure while still returning the block, enabling staged rollouts without breaking the node.
+  The underlying verifier bug fix from PR #2525 is already in place.
 # 2026-04-21
 - #2768 *Minor breaking change (code)*: Removed unused `Runtime::resolve_address` method from the native `Runtime` trait in `sov-modules-api`.
   The method had no call sites; address resolution continues to happen via `Accounts::resolve_sender_address{_read_only}` directly.

crates/adapters/celestia/src/config.rs

@@ -87,6 +87,10 @@ pub struct CelestiaConfig {
     /// Default: 30.
     #[serde(default = "default_background_stat_polling_interval_secs")]
     pub background_stat_polling_interval_secs: u64,
+    /// Controls how fetched Celestia blocks are verified against their Data Availability
+    /// Header before `get_block_at` returns. See [`VerifyOnFetchMode`].
+    #[serde(default)]
+    pub verify_on_fetch_mode: VerifyOnFetchMode,
     /// See [`sov_rollup_interface::node::da::DaService::safe_lead_time`].
     #[serde(default = "default_safe_lead_time_ms")]
     pub safe_lead_time_ms: u64,
@@ -137,6 +141,7 @@ impl fmt::Debug for CelestiaConfig {
                 "background_stat_polling_interval_secs",
                 &self.background_stat_polling_interval_secs,
             )
+            .field("verify_on_fetch_mode", &self.verify_on_fetch_mode)
             .field("safe_lead_time_ms", &self.safe_lead_time_ms)
             .field("tx_priority", &self.tx_priority)
             .field("backoff_min_delay_ms", &self.backoff_min_delay_ms)
@@ -165,6 +170,26 @@ impl From<TxPriority> for celestia_client::tx::TxPriority {
     }
 }
 
+/// Controls how fetched Celestia blocks are verified against their Data Availability Header
+/// before `get_block_at` returns.
+///
+/// Verification guards against silent consensus-breaking forks when the connected RPC node
+/// returns corrupted namespace data, at the cost of per-block overhead.
+#[derive(
+    Debug, Clone, Copy, Default, PartialEq, serde::Deserialize, serde::Serialize, JsonSchema,
+)]
+#[serde(rename_all = "snake_case")]
+pub enum VerifyOnFetchMode {
+    /// Skip verification entirely.
+    #[default]
+    Off,
+    /// Run verification; on failure log via `tracing::error!` and return the block anyway.
+    /// Recommended during staged rollouts when visibility matters more than hard enforcement.
+    LogError,
+    /// Run verification; on failure propagate the error so `get_block_at` fails.
+    ReturnError,
+}
+
 impl CelestiaConfig {
     /// Absolutely minimal config for client that is capable of reading
     pub fn minimal(rpc_url: String) -> Self {
@@ -179,6 +204,7 @@ impl CelestiaConfig {
             api_request_timeout_secs: default_api_request_timeout_secs(),
             tx_status_polling_millis: default_tx_status_polling_millis(),
             background_stat_polling_interval_secs: default_background_stat_polling_interval_secs(),
+            verify_on_fetch_mode: VerifyOnFetchMode::default(),
             safe_lead_time_ms: default_safe_lead_time_ms(),
             tx_priority: default_tx_priority(),
             backoff_min_delay_ms: default_min_delay_ms(),
@@ -335,7 +361,7 @@ pub(crate) fn default_background_stat_polling_interval_secs() -> u64 {
 
 #[cfg(test)]
 mod tests {
-    use super::{validate_rpc_url, CelestiaConfig, GrpcEndpointConfig};
+    use super::{validate_rpc_url, CelestiaConfig, GrpcEndpointConfig, VerifyOnFetchMode};
 
     const RPC_ENV_VAR: &str = "SOV_CELESTIA_RPC_URL";
     const GRPC_ENV_VAR: &str = "SOV_CELESTIA_GRPC_URL";
@@ -524,4 +550,35 @@ mod tests {
         let deserialized: CelestiaConfig = serde_json::from_str(&json).unwrap();
         assert_eq!(config, deserialized);
     }
+
+    #[test]
+    fn verify_on_fetch_mode_defaults_to_off() {
+        let _rpc_guard = EnvVarGuard::set(RPC_ENV_VAR, Some("ws://env-rpc:26658"));
+        let _grpc_guard = EnvVarGuard::set(GRPC_ENV_VAR, None);
+
+        let config = deserialize_config("{}").unwrap();
+        assert_eq!(config.verify_on_fetch_mode, VerifyOnFetchMode::Off);
+    }
+
+    #[test]
+    fn verify_on_fetch_mode_accepts_all_variants() {
+        let cases = [
+            (r#"{"verify_on_fetch_mode":"off"}"#, VerifyOnFetchMode::Off),
+            (
+                r#"{"verify_on_fetch_mode":"log_error"}"#,
+                VerifyOnFetchMode::LogError,
+            ),
+            (
+                r#"{"verify_on_fetch_mode":"return_error"}"#,
+                VerifyOnFetchMode::ReturnError,
+            ),
+        ];
+        for (json, expected) in cases {
+            let _rpc_guard = EnvVarGuard::set(RPC_ENV_VAR, Some("ws://env-rpc:26658"));
+            let _grpc_guard = EnvVarGuard::set(GRPC_ENV_VAR, None);
+
+            let config = deserialize_config(json).expect(json);
+            assert_eq!(config.verify_on_fetch_mode, expected);
+        }
+    }
 }

crates/adapters/celestia/src/da_service/mod.rs

@@ -1,7 +1,7 @@
 #[cfg(test)]
 mod tests;
 
-pub use crate::config::CelestiaConfig;
+pub use crate::config::{CelestiaConfig, VerifyOnFetchMode};
 use crate::metrics::client::{
     BlobGetAllMeasurement, GetBlockHeaderMeasurement, GetChainHeadMeasurement,
     GetNamespaceDataMeasurement, HeaderSyncStateMeasurement, StateBalanceForAddressMeasurement,
@@ -28,7 +28,9 @@ use celestia_types::nmt::Namespace;
 use futures::stream::BoxStream;
 use futures::StreamExt;
 use sov_rollup_interface::common::HexHash;
-use sov_rollup_interface::da::{DaProof, DaSpec, RelevantBlobs, RelevantProofs};
+use sov_rollup_interface::da::{
+    BlobReaderTrait, DaProof, DaSpec, DaVerifier, RelevantBlobs, RelevantProofs,
+};
 use sov_rollup_interface::node::da::{
     run_maybe_retryable_async_fn_with_retries, DaService, MaybeRetryable, SubmitBlobReceipt,
 };
@@ -45,6 +47,7 @@ pub struct CelestiaService {
     client: Arc<celestia_client::Client>,
     rollup_batch_namespace: Namespace,
     rollup_proof_namespace: Namespace,
+    verify_on_fetch_mode: VerifyOnFetchMode,
     signer_address: Option<CelestiaAddress>,
     safe_lead_time: Duration,
     backoff_policy: ExponentialBuilder,
@@ -54,10 +57,12 @@ pub struct CelestiaService {
 }
 
 impl CelestiaService {
+    #[allow(clippy::too_many_arguments)]
     fn with_client(
         client: celestia_client::Client,
         rollup_batch_namespace: Namespace,
         rollup_proof_namespace: Namespace,
+        verify_on_fetch_mode: VerifyOnFetchMode,
         signer_address: Option<CelestiaAddress>,
         safe_lead_time: Duration,
         backoff_policy: ExponentialBuilder,
@@ -69,6 +74,7 @@ impl CelestiaService {
             client: Arc::new(client),
             rollup_batch_namespace,
             rollup_proof_namespace,
+            verify_on_fetch_mode,
             signer_address,
             safe_lead_time,
             backoff_policy,
@@ -231,6 +237,7 @@ impl CelestiaService {
             client,
             chain_params.rollup_batch_namespace,
             chain_params.rollup_proof_namespace,
+            config.verify_on_fetch_mode,
             fetched_signer,
             Duration::from_millis(config.safe_lead_time_ms),
             backoff_policy,
@@ -356,7 +363,28 @@ impl CelestiaService {
             tracker.submit(get_block_measurement);
         });
         tracing::trace!(height, "get_block_at metrics send, returning");
-        FilteredCelestiaBlock::new(rollup_batch_shares, rollup_proof_shares, header)
+        let block = FilteredCelestiaBlock::new(rollup_batch_shares, rollup_proof_shares, header)?;
+        match self.verify_on_fetch_mode {
+            VerifyOnFetchMode::Off => {}
+            VerifyOnFetchMode::LogError => {
+                if let Err(error) = self.verify_block_integrity(&block) {
+                    tracing::error!(
+                        height,
+                        ?error,
+                        "Celestia block integrity verification failed; continuing because \
+                         verify_on_fetch_mode is `log_error`"
+                    );
+                }
+            }
+            VerifyOnFetchMode::ReturnError => {
+                self.verify_block_integrity(&block).map_err(|error| {
+                    anyhow::anyhow!(
+                        "Celestia block integrity verification failed at height {height}: {error}"
+                    )
+                })?;
+            }
+        }
+        Ok(block)
     }
 
     async fn get_head_block_header_inner(
@@ -423,6 +451,28 @@ impl CelestiaService {
             .map(|res| res.map(CelestiaHeader::from).map_err(|e| e.into()))
             .boxed())
     }
+
+    fn verify_block_integrity(&self, block: &FilteredCelestiaBlock) -> anyhow::Result<()> {
+        let verifier = CelestiaVerifier::new(RollupParams {
+            rollup_batch_namespace: self.rollup_batch_namespace,
+            rollup_proof_namespace: self.rollup_proof_namespace,
+        });
+
+        let mut relevant_blobs = extract_relevant_blobs(block);
+        // Advance full blob data first, then derive proofs for the consumed ranges.
+        for blob in relevant_blobs
+            .batch_blobs
+            .iter_mut()
+            .chain(relevant_blobs.proof_blobs.iter_mut())
+        {
+            blob.advance(blob.total_len());
+        }
+        let relevant_proofs = get_extraction_proof(block, &relevant_blobs);
+
+        verifier.verify_relevant_tx_list(&block.header, &relevant_blobs, relevant_proofs)?;
+
+        Ok(())
+    }
 }
 
 #[async_trait]

crates/adapters/celestia/src/lib.rs

@@ -20,6 +20,6 @@ pub use sov_rollup_interface::da::*;
 pub use sov_rollup_interface::node::da::*;
 
 #[cfg(feature = "native")]
-pub use da_service::{CelestiaConfig, CelestiaService};
+pub use da_service::{CelestiaConfig, CelestiaService, VerifyOnFetchMode};
 
 pub use crate::celestia::*;

crates/adapters/celestia/src/test_helper/docker.rs

@@ -9,7 +9,7 @@ use crate::config::{
     default_tx_status_polling_millis,
 };
 use crate::verifier::address::CelestiaAddress;
-use crate::{CelestiaConfig, CelestiaService};
+use crate::{CelestiaConfig, CelestiaService, VerifyOnFetchMode};
 use anyhow::{anyhow, Context};
 use sov_rollup_interface::da::BlockHeaderTrait;
 use sov_rollup_interface::node::da::DaService;
@@ -295,6 +295,7 @@ impl CelestiaDevNode {
             api_request_timeout_secs: default_api_request_timeout_secs(),
             tx_status_polling_millis: default_tx_status_polling_millis(),
             background_stat_polling_interval_secs: default_background_stat_polling_interval_secs(),
+            verify_on_fetch_mode: VerifyOnFetchMode::ReturnError,
             safe_lead_time_ms: default_safe_lead_time_ms(),
             tx_priority: default_tx_priority(),
             backoff_min_delay_ms: default_min_delay_ms(),

examples/demo-rollup/configs/celestia_rollup_config.toml

@@ -59,6 +59,13 @@ signer_private_key = "aec34bb1cfae6e906594dc106af4057a9046f769e2eeed67d040f0107e
 # Default: 30
 #background_stat_polling_interval_secs = 30
 
+# Controls how fetched Celestia blocks are verified against their Data Availability Header.
+# Options:
+#   "off"          - skip verification (fastest, may silently accept corrupted RPC data). Default.
+#   "log_error"    - run verification; on failure log and continue.
+#   "return_error" - run verification; on failure fail the fetch and propagate the error
+#verify_on_fetch_mode = "off"
+
 # Options are "Low", "Medium" and "High"
 # Defines a priority of rollup blob transactions. Set it to "High" for faster inclusion.
 # Default: High for better blob inclusion.