feat(conductor)!: implement chain ID checks

charts/evm-rollup/Chart.yaml

@@ -15,13 +15,13 @@ type: application
 # This is the chart version. This version number should be incremented each time you make changes
 # to the chart and its templates, including the app version.
 # Versions are expected to follow Semantic Versioning (https://semver.org/)
-version: 0.27.2
+version: 0.27.3
 
 # This is the version number of the application being deployed. This version number should be
 # incremented each time you make changes to the application. Versions are not expected to
 # follow Semantic Versioning. They should reflect the version the application is using.
 # It is recommended to use it with quotes.
-appVersion: "0.14.0"
+appVersion: "0.14.1"
 
 maintainers:
   - name: wafflesvonmaple

charts/evm-rollup/templates/configmap.yaml

@@ -31,6 +31,8 @@ data:
   OTEL_SERVICE_NAME: "{{ tpl .Values.otel.serviceNamePrefix . }}-conductor"
   {{- if not .Values.global.dev }}
   {{- else }}
+  ASTRIA_CONDUCTOR_SEQUENCER_CHAIN_ID: "{{ tpl .Values.config.conductor.sequencerChainId . }}"
+  ASTRIA_CONDUCTOR_CELESTIA_CHAIN_ID: "{{ tpl .Values.config.conductor.celestiaChainId . }}"
   {{- end }}
 ---
 apiVersion: v1

charts/evm-rollup/values.yaml

@@ -16,7 +16,7 @@ images:
   conductor:
     repo: ghcr.io/astriaorg/conductor
     pullPolicy: IfNotPresent
-    tag: "0.20.0"
+    tag: "0.20.1"
     devTag: latest
 
 
@@ -178,6 +178,8 @@ config:
     sequencerGrpc: ""
     # The maximum number of requests to make to the sequencer per second
     sequencerRequestsPerSecond: 500
+    # The chain id of the celestia network the conductor communicates with
+    celestiaChainId: ""
 
   celestia:
     # if config.rollup.executionLevel is NOT 'SoftOnly' AND celestia-node is not enabled

charts/evm-stack/Chart.lock

@@ -4,7 +4,7 @@ dependencies:
   version: 0.3.6
 - name: evm-rollup
   repository: file://../evm-rollup
-  version: 0.27.2
+  version: 0.27.3
 - name: composer
   repository: file://../composer
   version: 0.1.3
@@ -20,5 +20,5 @@ dependencies:
 - name: blockscout-stack
   repository: https://blockscout.github.io/helm-charts
   version: 1.6.2
-digest: sha256:7c27d6da529e8d1167805ece49047209936f51816b1d54c487e46c240dd4ed7a
-generated: "2024-09-12T10:11:18.397241-07:00"
+digest: sha256:0e67490b80b16f04e1e0ee5a1c5b904ff68084ecdd201ea74877e7e0e424da6f
+generated: "2024-09-16T14:33:09.843263-05:00"

charts/evm-stack/Chart.yaml

@@ -15,15 +15,15 @@ type: application
 # This is the chart version. This version number should be incremented each time you make changes
 # to the chart and its templates, including the app version.
 # Versions are expected to follow Semantic Versioning (https://semver.org/)
-version: 0.5.2
+version: 0.5.3
 
 dependencies:
   - name: celestia-node
     version: 0.3.6
     repository: "file://../celestia-node"
     condition: celestia-node.enabled
   - name: evm-rollup
-    version: 0.27.2
+    version: 0.27.3
     repository: "file://../evm-rollup"
   - name: composer
     version: 0.1.3

charts/evm-stack/values.yaml

@@ -13,6 +13,7 @@ global:
   rollupName: ""
   evmChainId: ""
   sequencerChainId: ""
+  celestiaChainId: ""
   otel:
     endpoint: ""
     tracesEndpoint: ""
@@ -29,6 +30,7 @@ evm-rollup:
   config:
     conductor:
       sequencerChainId: "{{ .Values.global.sequencerChainId }}"
+      celestiaChainId: "{{ .Values.global.celestiaChainId }}"
       sequencerRpc: "{{ .Values.global.sequencerRpc }}"
       sequencerGrpc: "{{ .Values.global.sequencerGrpc }}"
   otel:

crates/astria-conductor/local.env.example

@@ -73,6 +73,12 @@ ASTRIA_CONDUCTOR_SEQUENCER_BLOCK_TIME_MS=2000
 # CometBFT node.
 ASTRIA_CONDUCTOR_SEQUENCER_REQUESTS_PER_SECOND=500
 
+# The chain ID of the sequencer network the conductor should be communicating with.
+ASTRIA_CONDUCTOR_SEQUENCER_CHAIN_ID="test-sequencer-1000"
+
+# The chain ID of the Celestia network the conductor should be communicating with.
+ASTRIA_CONDUCTOR_CELESTIA_CHAIN_ID="test-celestia-1000"
+
 # Set to true to enable prometheus metrics.
 ASTRIA_CONDUCTOR_NO_METRICS=true
 

crates/astria-conductor/src/celestia/builder.rs

@@ -23,6 +23,7 @@ pub(crate) struct Builder {
     pub(crate) executor: executor::Handle,
     pub(crate) sequencer_cometbft_client: SequencerClient,
     pub(crate) sequencer_requests_per_second: u32,
+    pub(crate) celestia_chain_id: String,
     pub(crate) shutdown: CancellationToken,
     pub(crate) metrics: &'static Metrics,
 }
@@ -37,6 +38,7 @@ impl Builder {
             executor,
             sequencer_cometbft_client,
             sequencer_requests_per_second,
+            celestia_chain_id,
             shutdown,
             metrics,
         } = self;
@@ -50,6 +52,7 @@ impl Builder {
             executor,
             sequencer_cometbft_client,
             sequencer_requests_per_second,
+            celestia_chain_id,
             shutdown,
             metrics,
         })

crates/astria-conductor/src/celestia/mod.rs

@@ -11,9 +11,11 @@ use astria_core::{
 use astria_eyre::eyre::{
     self,
     bail,
+    ensure,
     WrapErr as _,
 };
 use bytes::Bytes;
+use celestia_rpc::HeaderClient as _;
 use celestia_types::nmt::Namespace;
 use futures::{
     future::{
@@ -95,6 +97,14 @@ use crate::{
     executor,
 };
 
+#[derive(Debug, thiserror::Error)]
+pub enum ValidateChainIdError {
+    #[error("failed to get the Celestia network head")]
+    GetCelestiaNetworkHead(#[source] jsonrpsee::core::Error),
+    #[error("expected Celestia chain id `{expected}` does not match actual: `{actual}`")]
+    MismatchedCelestiaChainId { expected: String, actual: String },
+}
+
 /// Sequencer Block information reconstructed from Celestia blobs.
 ///
 /// Will be forwarded to the executor as a firm block.
@@ -139,6 +149,9 @@ pub(crate) struct Reader {
     /// (usually to verify block data retrieved from Celestia blobs).
     sequencer_requests_per_second: u32,
 
+    /// The chain ID of the Celestia network the reader should be communicating with.
+    celestia_chain_id: String,
+
     /// Token to listen for Conductor being shut down.
     shutdown: CancellationToken,
 
@@ -170,6 +183,17 @@ impl Reader {
     async fn initialize(
         &mut self,
     ) -> eyre::Result<(executor::Handle<StateIsInit>, tendermint::chain::Id)> {
+        let actual_chain_id = get_celestia_chain_id(&self.celestia_client)
+            .await
+            .wrap_err("failed to fetch Celestia chain ID")?;
+        ensure!(
+            self.celestia_chain_id == actual_chain_id,
+            ValidateChainIdError::MismatchedCelestiaChainId {
+                expected: self.celestia_chain_id.clone(),
+                actual: actual_chain_id,
+            }
+        );
+
         let wait_for_init_executor = async {
             self.executor
                 .wait_for_init()
@@ -187,6 +211,34 @@ impl Reader {
     }
 }
 
+#[instrument(skip_all, err)]
+async fn get_celestia_chain_id(
+    celestia_client: &CelestiaClient,
+) -> Result<String, ValidateChainIdError> {
+    let retry_config = tryhard::RetryFutureConfig::new(u32::MAX)
+        .exponential_backoff(Duration::from_millis(100))
+        .max_delay(Duration::from_secs(20))
+        .on_retry(
+            |attempt: u32, next_delay: Option<Duration>, error: &jsonrpsee::core::Error| {
+                let wait_duration = next_delay
+                    .map(humantime::format_duration)
+                    .map(tracing::field::display);
+                warn!(
+                    attempt,
+                    wait_duration,
+                    error = error as &dyn std::error::Error,
+                    "attempt to fetch celestia network header info; retrying after backoff",
+                );
+                futures::future::ready(())
+            },
+        );
+    let network_head = tryhard::retry_fn(|| celestia_client.header_network_head())
+        .with_config(retry_config)
+        .await
+        .map_err(ValidateChainIdError::GetCelestiaNetworkHead)?;
+    Ok(network_head.chain_id().to_string())
+}
+
 struct RunningReader {
     block_cache: BlockCache<ReconstructedBlock>,
 
@@ -610,7 +662,9 @@ async fn enqueue_block(
 }
 
 #[instrument(skip_all, err)]
-async fn get_sequencer_chain_id(client: SequencerClient) -> eyre::Result<tendermint::chain::Id> {
+pub(crate) async fn get_sequencer_chain_id(
+    client: SequencerClient,
+) -> eyre::Result<tendermint::chain::Id> {
     use sequencer_client::Client as _;
 
     let retry_config = tryhard::RetryFutureConfig::new(u32::MAX)

crates/astria-conductor/src/conductor.rs

@@ -8,8 +8,10 @@ use astria_eyre::eyre::{
     eyre,
     WrapErr as _,
 };
+pub use celestia::ValidateChainIdError as CelestiaChainIdError;
 use itertools::Itertools as _;
 use pin_project_lite::pin_project;
+pub use sequencer::ValidateChainIdError as SequencerChainIdError;
 use sequencer_client::HttpClient;
 use tokio::{
     select,
@@ -40,7 +42,7 @@ pin_project! {
     /// A handle returned by [`Conductor::spawn`].
     pub struct Handle {
         shutdown_token: CancellationToken,
-        task: Option<tokio::task::JoinHandle<()>>,
+        task: Option<tokio::task::JoinHandle<eyre::Result<()>>>,
     }
 }
 
@@ -53,15 +55,15 @@ impl Handle {
     /// # Panics
     /// Panics if called twice.
     #[instrument(skip_all, err)]
-    pub async fn shutdown(&mut self) -> Result<(), tokio::task::JoinError> {
+    pub async fn shutdown(&mut self) -> Result<eyre::Result<()>, tokio::task::JoinError> {
         self.shutdown_token.cancel();
         let task = self.task.take().expect("shutdown must not be called twice");
         task.await
     }
 }
 
 impl Future for Handle {
-    type Output = Result<(), tokio::task::JoinError>;
+    type Output = Result<eyre::Result<()>, tokio::task::JoinError>;
 
     fn poll(
         self: std::pin::Pin<&mut Self>,
@@ -131,6 +133,7 @@ impl Conductor {
                 sequencer_grpc_client,
                 sequencer_cometbft_client: sequencer_cometbft_client.clone(),
                 sequencer_block_time: Duration::from_millis(cfg.sequencer_block_time_ms),
+                sequencer_chain_id: cfg.sequencer_chain_id,
                 shutdown: shutdown.clone(),
                 executor: executor_handle.clone(),
             }
@@ -152,6 +155,7 @@ impl Conductor {
                 executor: executor_handle.clone(),
                 sequencer_cometbft_client: sequencer_cometbft_client.clone(),
                 sequencer_requests_per_second: cfg.sequencer_requests_per_second,
+                celestia_chain_id: cfg.celestia_chain_id,
                 shutdown: shutdown.clone(),
                 metrics,
             }
@@ -171,7 +175,7 @@ impl Conductor {
     ///
     /// # Panics
     /// Panics if it could not install a signal handler.
-    async fn run_until_stopped(mut self) {
+    async fn run_until_stopped(mut self) -> eyre::Result<()> {
         info_span!("Conductor::run_until_stopped").in_scope(|| info!("conductor is running"));
 
         let exit_reason = select! {
@@ -184,14 +188,15 @@ impl Conductor {
             Some((name, res)) = self.tasks.join_next() => {
                 match flatten(res) {
                     Ok(()) => Err(eyre!("task `{name}` exited unexpectedly")),
-                    Err(err) => Err(err).wrap_err_with(|| "task `{name}` failed"),
+                    Err(err) => Err(err).wrap_err_with(|| "task `{name}` failed")?,
                 }
             }
         };
 
         let message = "initiating shutdown";
         report_exit(exit_reason, message);
         self.shutdown().await;
+        Ok(())
     }
 
     /// Spawns Conductor on the tokio runtime.

crates/astria-conductor/src/config.rs

@@ -61,6 +61,12 @@ pub struct Config {
     /// The number of requests per second that will be sent to Sequencer.
     pub sequencer_requests_per_second: u32,
 
+    /// The chain ID of the sequencer network the conductor should be communiacting with.
+    pub sequencer_chain_id: String,
+
+    /// The chain ID of the Celestia network the conductor should be communicating with.
+    pub celestia_chain_id: String,
+
     /// Address of the RPC server for execution
     pub execution_rpc_url: String,
 

crates/astria-conductor/src/sequencer/builder.rs

@@ -10,6 +10,7 @@ pub(crate) struct Builder {
     pub(crate) sequencer_grpc_client: SequencerGrpcClient,
     pub(crate) sequencer_cometbft_client: sequencer_client::HttpClient,
     pub(crate) sequencer_block_time: Duration,
+    pub(crate) sequencer_chain_id: String,
     pub(crate) shutdown: CancellationToken,
 }
 
@@ -20,13 +21,15 @@ impl Builder {
             sequencer_grpc_client,
             sequencer_cometbft_client,
             sequencer_block_time,
+            sequencer_chain_id,
             shutdown,
         } = self;
         super::Reader {
             executor,
             sequencer_grpc_client,
             sequencer_cometbft_client,
             sequencer_block_time,
+            sequencer_chain_id,
             shutdown,
         }
     }