feat(sequencer-relayer)!: add chain IDs for sequencer and Celestia to config env vars

charts/sequencer-relayer/Chart.yaml

@@ -15,7 +15,7 @@ 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.8.3
+version: 0.8.4
 
 # 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

charts/sequencer-relayer/templates/configmaps.yaml

@@ -30,6 +30,8 @@ data:
   ASTRIA_SEQUENCER_RELAYER_ONLY_INCLUDE_ROLLUPS: "{{ .Values.config.relayer.onlyIncludeRollups }}"
   {{- if not .Values.global.dev }}
   {{- else }}
+  ASTRIA_SEQUENCER_RELAYER_SEQUENCER_CHAIN_ID: "{{ .Values.config.relayer.sequencerChainId }}"
+  ASTRIA_SEQUENCER_RELAYER_CELESTIA_CHAIN_ID: "{{ .Values.config.relayer.celestiaChainId }}"
   {{- end }}
 ---
 apiVersion: v1

charts/sequencer-relayer/values.yaml

@@ -18,6 +18,8 @@ images:
 
 config:
   relayer:
+    sequencerChainId: ""
+    celestiaChainId: ""
     celestiaAppGrpc: ""
     cometbftRpc: ""
     sequencerGrpc: ""

charts/sequencer/Chart.lock

@@ -1,6 +1,6 @@
 dependencies:
 - name: sequencer-relayer
   repository: file://../sequencer-relayer
-  version: 0.8.3
-digest: sha256:5299fcd71b8cac0aa3929a00830fa682d04d239700664a644b3010980704e867
-generated: "2024-05-28T13:16:38.413956-07:00"
+  version: 0.8.4
+digest: sha256:fa980cbe23353b162aad741afbf2ccde4d93b6d017be5fe478d5994bcfd1c992
+generated: "2024-05-30T17:09:44.161456347+01:00"

charts/sequencer/Chart.yaml

@@ -15,7 +15,7 @@ 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.15.3
+version: 0.15.4
 
 # 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
@@ -25,7 +25,7 @@ appVersion: "0.13.0"
 
 dependencies:
   - name: sequencer-relayer
-    version: "0.8.3"
+    version: "0.8.4"
     repository: "file://../sequencer-relayer"
     condition: sequencer-relayer.enabled
 

crates/astria-sequencer-relayer/local.env.example

@@ -19,6 +19,12 @@ ASTRIA_SEQUENCER_RELAYER_PRETTY_PRINT=false
 # `ASTRIA_SEQUENCER_RELAYER_PRETTY_PRINT` is set to `true`.
 NO_COLOR=
 
+# The chain ID of the sequencer network from which this sequencer-relayer will fetch blocks.
+ASTRIA_SEQUENCER_RELAYER_SEQUENCER_CHAIN_ID="sequencer-test-chain-0"
+
+# The chain ID of the Celestia network to which this sequencer-relayer will submit data.
+ASTRIA_SEQUENCER_RELAYER_CELESTIA_CHAIN_ID="celestia-local-0"
+
 # Address of cometbft/tendermint to request new block heights.
 # 127.0.0.1:26657 is the default socket address at which cometbft
 # serves RPCs.

crates/astria-sequencer-relayer/src/config.rs

@@ -24,6 +24,8 @@ use serde::{
 #[derive(Clone, Debug, Deserialize, Serialize, PartialEq)]
 /// The single config for creating an astria-sequencer-relayer service.
 pub struct Config {
+    pub sequencer_chain_id: String,
+    pub celestia_chain_id: String,
     pub cometbft_endpoint: String,
     pub sequencer_grpc_endpoint: String,
     pub celestia_app_grpc_endpoint: String,

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

@@ -27,6 +27,8 @@ use crate::{
 
 pub(crate) struct Builder {
     pub(crate) shutdown_token: tokio_util::sync::CancellationToken,
+    pub(crate) sequencer_chain_id: String,
+    pub(crate) celestia_chain_id: String,
     pub(crate) celestia_app_grpc_endpoint: String,
     pub(crate) celestia_app_key_file: String,
     pub(crate) cometbft_endpoint: String,
@@ -43,6 +45,8 @@ impl Builder {
     pub(crate) fn build(self) -> eyre::Result<super::Relayer> {
         let Self {
             shutdown_token,
+            sequencer_chain_id,
+            celestia_chain_id,
             celestia_app_grpc_endpoint,
             celestia_app_key_file,
             cometbft_endpoint,
@@ -77,12 +81,13 @@ impl Builder {
                 .wrap_err("failed parsing provided celestia app grpc endpoint as Uri")?;
             let celestia_keys = CelestiaKeys::from_path(celestia_app_key_file)
                 .wrap_err("failed to get celestia keys from file")?;
-            CelestiaClientBuilder::new(uri, celestia_keys, state.clone())
+            CelestiaClientBuilder::new(celestia_chain_id, uri, celestia_keys, state.clone())
                 .wrap_err("failed to create celestia client builder")?
         };
 
         Ok(super::Relayer {
             shutdown_token,
+            sequencer_chain_id,
             sequencer_cometbft_client,
             sequencer_grpc_client,
             sequencer_poll_period,

crates/astria-sequencer-relayer/src/relayer/celestia_client/builder.rs

@@ -14,7 +14,10 @@ use tonic::transport::{
     Channel,
     Endpoint,
 };
-use tracing::trace;
+use tracing::{
+    info,
+    trace,
+};
 
 use super::{
     super::State,
@@ -40,6 +43,14 @@ pub(in crate::relayer) enum BuilderError {
     /// The node info response was empty.
     #[error("the celestia node info response was empty")]
     EmptyNodeInfo,
+    /// Mismatch in Celestia chain ID.
+    #[error(
+        "mismatch in celestia chain id, configured id: `{configured}`, received id: `{received}`"
+    )]
+    MismatchedCelestiaChainId {
+        configured: String,
+        received: String,
+    },
 }
 
 /// An error while encoding a Bech32 string.
@@ -50,6 +61,7 @@ pub(in crate::relayer) struct Bech32EncodeError(#[from] bech32::EncodeError);
 /// A builder for a [`CelestiaClient`].
 #[derive(Clone)]
 pub(in crate::relayer) struct Builder {
+    configured_celestia_chain_id: String,
     /// The inner `tonic` gRPC channel shared by the various generated gRPC clients.
     grpc_channel: Channel,
     /// The crypto keys associated with our Celestia account.
@@ -63,13 +75,15 @@ pub(in crate::relayer) struct Builder {
 impl Builder {
     /// Returns a new `Builder`, or an error if Bech32-encoding the `signing_keys` address fails.
     pub(in crate::relayer) fn new(
+        configured_celestia_chain_id: String,
         uri: Uri,
         signing_keys: CelestiaKeys,
         state: Arc<State>,
     ) -> Result<Self, BuilderError> {
         let grpc_channel = Endpoint::from(uri).connect_lazy();
         let address = bech32_encode(&signing_keys.address)?;
         Ok(Self {
+            configured_celestia_chain_id,
             grpc_channel,
             signing_keys,
             address,
@@ -79,14 +93,24 @@ impl Builder {
 
     /// Returns a new `CelestiaClient` initialized with info retrieved from the Celestia app.
     pub(in crate::relayer) async fn try_build(self) -> Result<CelestiaClient, BuilderError> {
-        let chain_id = self.fetch_chain_id().await?;
+        let reeceived_celestia_chain_id = self.fetch_celestia_chain_id().await?;
 
         let Self {
+            configured_celestia_chain_id,
             grpc_channel,
             signing_keys,
             address,
             state,
         } = self;
+
+        if reeceived_celestia_chain_id != configured_celestia_chain_id {
+            return Err(BuilderError::MismatchedCelestiaChainId {
+                configured: configured_celestia_chain_id,
+                received: reeceived_celestia_chain_id,
+            });
+        }
+
+        info!(celestia_chain_id = %reeceived_celestia_chain_id, "confirmed celestia chain id");
         state.set_celestia_connected(true);
 
         let tx_client = TxClient::new(grpc_channel.clone());
@@ -95,11 +119,11 @@ impl Builder {
             tx_client,
             signing_keys,
             address,
-            chain_id,
+            chain_id: reeceived_celestia_chain_id,
         })
     }
 
-    async fn fetch_chain_id(&self) -> Result<String, BuilderError> {
+    async fn fetch_celestia_chain_id(&self) -> Result<String, BuilderError> {
         let mut node_info_client = NodeInfoClient::new(self.grpc_channel.clone());
         let response = node_info_client.get_node_info(GetNodeInfoRequest {}).await;
         // trace-level logging, so using Debug format is ok.

crates/astria-sequencer-relayer/src/relayer/mod.rs

@@ -14,6 +14,7 @@ use astria_core::{
 use astria_eyre::eyre::{
     self,
     bail,
+    ensure,
     eyre,
     WrapErr as _,
 };
@@ -27,6 +28,7 @@ use futures::{
 };
 use sequencer_client::{
     tendermint::block::Height as SequencerHeight,
+    tendermint_rpc,
     HttpClient as SequencerClient,
 };
 use tokio::{
@@ -46,7 +48,10 @@ use tracing::{
     field::DisplayValue,
     info,
     instrument,
+    trace,
     warn,
+    Instrument,
+    Span,
 };
 
 use crate::{
@@ -77,6 +82,9 @@ pub(crate) struct Relayer {
     /// A token to notify relayer that it should shut down.
     shutdown_token: CancellationToken,
 
+    /// The configured chain ID of the sequencer network.
+    sequencer_chain_id: String,
+
     /// The client used to query the sequencer cometbft endpoint.
     sequencer_cometbft_client: SequencerClient,
 
@@ -119,6 +127,14 @@ impl Relayer {
             .await
             .wrap_err("failed reading submission state from files")?;
 
+        select!(
+            () = self.shutdown_token.cancelled() => return Ok(()),
+            init_result = confirm_sequencer_chain_id(
+                self.sequencer_chain_id.clone(),
+                self.sequencer_cometbft_client.clone()
+            ) => init_result,
+        )?;
+
         let last_submitted_sequencer_height = submission_state.last_submitted_height();
 
         let mut latest_height_stream = {
@@ -288,6 +304,62 @@ impl Relayer {
     }
 }
 
+#[instrument(skip_all)]
+async fn confirm_sequencer_chain_id(
+    configured_sequencer_chain_id: String,
+    sequencer_cometbft_client: SequencerClient,
+) -> eyre::Result<()> {
+    let span = Span::current();
+
+    let retry_config = tryhard::RetryFutureConfig::new(u32::MAX)
+        .max_delay(Duration::from_secs(30))
+        .exponential_backoff(Duration::from_secs(1))
+        .on_retry(
+            |attempt: u32, next_delay: Option<Duration>, error: &tendermint_rpc::Error| {
+                let wait_duration = next_delay
+                    .map(humantime::format_duration)
+                    .map(tracing::field::display);
+                warn!(
+                    parent: &span,
+                    attempt,
+                    wait_duration,
+                    error = %eyre::Report::new(error.clone()),
+                    "failed to fetch sequencer chain id; retrying after backoff",
+                );
+                futures::future::ready(())
+            },
+        );
+
+    let received_sequencer_chain_id =
+        tryhard::retry_fn(move || fetch_sequencer_chain_id(sequencer_cometbft_client.clone()))
+            .with_config(retry_config)
+            .in_current_span()
+            .await
+            .wrap_err("retry attempts exhausted; bailing")?;
+
+    ensure!(
+        received_sequencer_chain_id == configured_sequencer_chain_id,
+        "configured sequencer chain ID does not match received; configured: \
+         `{configured_sequencer_chain_id}`, received: `{received_sequencer_chain_id}`"
+    );
+    info!(sequencer_chain_id = %configured_sequencer_chain_id, "confirmed sequencer chain id");
+    Ok(())
+}
+
+async fn fetch_sequencer_chain_id(
+    sequencer_cometbft_client: SequencerClient,
+) -> Result<String, tendermint_rpc::Error> {
+    use sequencer_client::Client as _;
+
+    let response = sequencer_cometbft_client.status().await;
+    // trace-level logging, so using Debug format is ok.
+    #[cfg_attr(dylint_lib = "tracing_debug_field", allow(tracing_debug_field))]
+    {
+        trace!(?response);
+    }
+    response.map(|status_response| status_response.node_info.network.to_string())
+}
+
 async fn read_submission_state<P1: AsRef<Path>, P2: AsRef<Path>>(
     pre: P1,
     post: P2,

crates/astria-sequencer-relayer/src/relayer/write/mod.rs

@@ -156,6 +156,7 @@ impl BlobSubmitter {
         let client = init_result.map_err(|error| {
             let message = "failed to initialize celestia client";
             error!(%error, message);
+            self.shutdown_token.cancel();
             error.wrap_err(message)
         })?;
 
@@ -355,9 +356,20 @@ async fn submit_blobs(
 async fn init_with_retry(client_builder: CelestiaClientBuilder) -> eyre::Result<CelestiaClient> {
     let span = Span::current();
 
+    let initial_retry_delay = Duration::from_secs(1);
     let retry_config = tryhard::RetryFutureConfig::new(u32::MAX)
-        .exponential_backoff(Duration::from_secs(1))
         .max_delay(Duration::from_secs(30))
+        .custom_backoff(|attempt: u32, error: &BuilderError| {
+            if matches!(error, BuilderError::MismatchedCelestiaChainId { .. }) {
+                // We got a good response from the Celestia app, but this is an unrecoverable error.
+                return tryhard::RetryPolicy::Break;
+            }
+            // This is equivalent to the `exponential_backoff` policy.  Note that `max_delay`
+            // above is still respected regardless of what we return here.
+            let delay =
+                initial_retry_delay.saturating_mul(2_u32.saturating_pow(attempt.saturating_sub(1)));
+            tryhard::RetryPolicy::Delay(delay)
+        })
         .on_retry(
             |attempt: u32, next_delay: Option<Duration>, error: &BuilderError| {
                 let wait_duration = next_delay
@@ -378,7 +390,7 @@ async fn init_with_retry(client_builder: CelestiaClientBuilder) -> eyre::Result<
         .with_config(retry_config)
         .in_current_span()
         .await
-        .wrap_err("retry attempts exhausted; bailing")?;
+        .wrap_err("failed to initialize celestia client")?;
     info!("initialized celestia client");
     Ok(celestia_client)
 }

crates/astria-sequencer-relayer/src/sequencer_relayer.rs

@@ -47,6 +47,8 @@ impl SequencerRelayer {
         let shutdown_handle = ShutdownHandle::new();
         let rollup_filter = cfg.only_include_rollups()?;
         let Config {
+            sequencer_chain_id,
+            celestia_chain_id,
             cometbft_endpoint,
             sequencer_grpc_endpoint,
             celestia_app_grpc_endpoint,
@@ -63,6 +65,8 @@ impl SequencerRelayer {
         let validator_key_path = relay_only_validator_key_blocks.then_some(validator_key_file);
         let relayer = relayer::Builder {
             shutdown_token: shutdown_handle.token(),
+            sequencer_chain_id,
+            celestia_chain_id,
             celestia_app_grpc_endpoint,
             celestia_app_key_file,
             cometbft_endpoint,