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_EXPECTED_SEQUENCER_CHAIN_ID: "{{ tpl .Values.config.conductor.sequencerChainId . }}"
+  ASTRIA_CONDUCTOR_EXPECTED_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_EXPECTED_SEQUENCER_CHAIN_ID="test-sequencer-1000"
+
+# The chain ID of the Celestia network the conductor should be communicating with.
+ASTRIA_CONDUCTOR_EXPECTED_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) expected_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,
+            expected_celestia_chain_id,
             shutdown,
             metrics,
         } = self;
@@ -50,6 +52,7 @@ impl Builder {
             executor,
             sequencer_cometbft_client,
             sequencer_requests_per_second,
+            expected_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::{
@@ -139,6 +141,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.
+    expected_celestia_chain_id: String,
+
     /// Token to listen for Conductor being shut down.
     shutdown: CancellationToken,
 
@@ -170,6 +175,16 @@ impl Reader {
     async fn initialize(
         &mut self,
     ) -> eyre::Result<(executor::Handle<StateIsInit>, tendermint::chain::Id)> {
+        let actual_celestia_chain_id = get_celestia_chain_id(&self.celestia_client)
+            .await
+            .wrap_err("failed to fetch Celestia chain ID")?;
+        let expected_celestia_chain_id = &self.expected_celestia_chain_id;
+        ensure!(
+            self.expected_celestia_chain_id == actual_celestia_chain_id,
+            "expected Celestia chain id `{expected_celestia_chain_id}` does not match actual: \
+             `{actual_celestia_chain_id}`"
+        );
+
         let wait_for_init_executor = async {
             self.executor
                 .wait_for_init()
@@ -187,6 +202,31 @@ impl Reader {
     }
 }
 
+#[instrument(skip_all, err)]
+async fn get_celestia_chain_id(celestia_client: &CelestiaClient) -> eyre::Result<String> {
+    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?;
+    Ok(network_head.chain_id().to_string())
+}
+
 struct RunningReader {
     block_cache: BlockCache<ReconstructedBlock>,
 
@@ -610,7 +650,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/inner.rs

@@ -6,6 +6,7 @@ use std::{
 use astria_eyre::eyre::{
     self,
     eyre,
+    Result,
     WrapErr as _,
 };
 use itertools::Itertools as _;
@@ -55,12 +56,12 @@ pin_project! {
     /// A handle returned by [`ConductorInner::spawn`].
     pub(super) struct InnerHandle {
         shutdown_token: CancellationToken,
-        task: Option<tokio::task::JoinHandle<RestartOrShutdown>>,
+        task: Option<tokio::task::JoinHandle<Result<RestartOrShutdown>>>,
     }
 }
 
 impl Future for InnerHandle {
-    type Output = Result<RestartOrShutdown, tokio::task::JoinError>;
+    type Output = Result<Result<RestartOrShutdown>, tokio::task::JoinError>;
 
     fn poll(
         self: std::pin::Pin<&mut Self>,
@@ -132,6 +133,7 @@ impl ConductorInner {
                 sequencer_grpc_client,
                 sequencer_cometbft_client: sequencer_cometbft_client.clone(),
                 sequencer_block_time: Duration::from_millis(cfg.sequencer_block_time_ms),
+                expected_sequencer_chain_id: cfg.expected_sequencer_chain_id,
                 shutdown: shutdown_token.clone(),
                 executor: executor_handle.clone(),
             }
@@ -153,6 +155,7 @@ impl ConductorInner {
                 executor: executor_handle.clone(),
                 sequencer_cometbft_client: sequencer_cometbft_client.clone(),
                 sequencer_requests_per_second: cfg.sequencer_requests_per_second,
+                expected_celestia_chain_id: cfg.expected_celestia_chain_id,
                 shutdown: shutdown_token.clone(),
                 metrics,
             }
@@ -172,7 +175,7 @@ impl ConductorInner {
     ///
     /// # Panics
     /// Panics if it could not install a signal handler.
-    async fn run_until_stopped(mut self) -> RestartOrShutdown {
+    async fn run_until_stopped(mut self) -> Result<RestartOrShutdown> {
         info_span!("Conductor::run_until_stopped").in_scope(|| info!("conductor is running"));
 
         let exit_reason = select! {
@@ -219,7 +222,7 @@ impl ConductorInner {
     /// because kubernetes issues SIGKILL 30 seconds after SIGTERM, giving 5 seconds
     /// to abort the remaining tasks.
     #[instrument(skip_all)]
-    async fn shutdown(mut self, exit_reason: ExitReason) -> RestartOrShutdown {
+    async fn shutdown(mut self, exit_reason: ExitReason) -> Result<RestartOrShutdown> {
         self.shutdown_token.cancel();
         let mut restart_or_shutdown = RestartOrShutdown::Shutdown;
 
@@ -273,7 +276,15 @@ impl ConductorInner {
         }
         info!("shutting down");
 
-        restart_or_shutdown
+        if let ExitReason::TaskFailed {
+            error, ..
+        } = exit_reason
+        {
+            if matches!(restart_or_shutdown, RestartOrShutdown::Shutdown) {
+                return Err(error);
+            }
+        }
+        Ok(restart_or_shutdown)
     }
 }
 

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

@@ -2,7 +2,10 @@ mod inner;
 
 use std::future::Future;
 
-use astria_eyre::eyre;
+use astria_eyre::eyre::{
+    self,
+    Result,
+};
 use inner::{
     ConductorInner,
     InnerHandle,
@@ -128,16 +131,17 @@ impl Conductor {
     #[instrument(skip_all, err)]
     async fn shutdown_or_restart(
         &mut self,
-        exit_reason: Result<RestartOrShutdown, JoinError>,
+        exit_reason: Result<Result<RestartOrShutdown>, JoinError>,
     ) -> eyre::Result<&'static str> {
         match exit_reason {
-            Ok(restart_or_shutdown) => match restart_or_shutdown {
+            Ok(Ok(restart_or_shutdown)) => match restart_or_shutdown {
                 RestartOrShutdown::Restart => {
                     self.restart();
                     return Ok("restarting");
                 }
                 RestartOrShutdown::Shutdown => Ok("conductor exiting"),
             },
+            Ok(Err(err)) => Err(err.wrap_err("conductor failed failed")),
             Err(err) => Err(eyre::ErrReport::from(err).wrap_err("conductor failed")),
         }
     }

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 expected_sequencer_chain_id: String,
+
+    /// The chain ID of the Celestia network the conductor should be communicating with.
+    pub expected_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) expected_sequencer_chain_id: String,
     pub(crate) shutdown: CancellationToken,
 }
 
@@ -20,13 +21,15 @@ impl Builder {
             sequencer_grpc_client,
             sequencer_cometbft_client,
             sequencer_block_time,
+            expected_sequencer_chain_id,
             shutdown,
         } = self;
         super::Reader {
             executor,
             sequencer_grpc_client,
             sequencer_cometbft_client,
             sequencer_block_time,
+            expected_sequencer_chain_id,
             shutdown,
         }
     }