feat(composer)!: Add chain_id check on executor build

crates/astria-composer/src/executor/builder.rs

@@ -15,7 +15,7 @@ use astria_core::{
 use astria_eyre::eyre::{
     self,
     eyre,
-    WrapErr as _,
+    Context,
 };
 use tokio::sync::watch;
 use tokio_util::sync::CancellationToken;
@@ -48,6 +48,7 @@ impl Builder {
         } = self;
         let sequencer_client = sequencer_client::HttpClient::new(sequencer_url.as_str())
             .wrap_err("failed constructing sequencer client")?;
+
         let (status, _) = watch::channel(Status::new());
 
         let sequencer_key = read_signing_key_from_file(&private_key_file).wrap_err_with(|| {

crates/astria-composer/src/executor/mod.rs

@@ -24,6 +24,7 @@ use astria_core::{
 };
 use astria_eyre::eyre::{
     self,
+    ensure,
     eyre,
     WrapErr as _,
 };
@@ -40,7 +41,10 @@ use futures::{
 use pin_project_lite::pin_project;
 use prost::Message as _;
 use sequencer_client::{
-    tendermint_rpc::endpoint::broadcast::tx_sync,
+    tendermint_rpc::{
+        endpoint::broadcast::tx_sync,
+        Client,
+    },
     Address,
     SequencerClientExt as _,
 };
@@ -192,6 +196,10 @@ impl Executor {
     /// An error is returned if connecting to the sequencer fails.
     #[instrument(skip_all, fields(address = %self.address))]
     pub(super) async fn run_until_stopped(mut self) -> eyre::Result<()> {
+        let _chain_id_result = self
+            .ensure_configured_chain_id_matches_remote()
+            .await
+            .wrap_err("failed to validate chain_id")?;
         let mut submission_fut: Fuse<Instrumented<SubmitFut>> = Fuse::terminated();
         let mut nonce = get_latest_nonce(self.sequencer_client.clone(), self.address)
             .await
@@ -415,6 +423,41 @@ impl Executor {
 
         reason.map(|_| ())
     }
+
+    // check for mismatched configured chain_id and sequencer client chain_id
+    pub(crate) async fn ensure_configured_chain_id_matches_remote(&self) -> eyre::Result<()> {
+        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: &sequencer_client::tendermint_rpc::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 sequencer genesis info; retrying after backoff",
+                    );
+                    futures::future::ready(())
+                },
+            );
+        let client_genesis: tendermint::Genesis =
+            tryhard::retry_fn(|| self.sequencer_client.genesis())
+                .with_config(retry_config)
+                .await
+                .wrap_err("failed to retrieve sequencer genesis after many attempts")?;
+        ensure!(
+            self.sequencer_chain_id == client_genesis.chain_id.as_str(),
+            "mismatch in configured chain_id: {0} and sequencer chain_id: {1}",
+            self.sequencer_chain_id,
+            client_genesis.chain_id.as_str()
+        );
+        Ok(())
+    }
 }
 
 /// Queries the sequencer for the latest nonce with an exponential backoff

crates/astria-composer/src/executor/tests.rs

@@ -4,6 +4,7 @@ use std::{
 };
 
 use astria_core::{
+    generated::protocol::account::v1alpha1::NonceResponse,
     primitive::v1::{
         asset::default_native_asset,
         RollupId,
@@ -18,6 +19,17 @@ use prost::Message;
 use sequencer_client::SignedTransaction;
 use serde_json::json;
 use tempfile::NamedTempFile;
+use tendermint::{
+    consensus::{
+        params::{
+            AbciParams,
+            ValidatorParams,
+        },
+        Params,
+    },
+    Genesis,
+    Time,
+};
 use tendermint_rpc::{
     endpoint::broadcast::tx_sync,
     request,
@@ -65,20 +77,10 @@ static TELEMETRY: Lazy<()> = Lazy::new(|| {
     }
 });
 
-/// Start a mock sequencer server and mount a mock for the `accounts/nonce` query.
-async fn setup() -> (MockServer, MockGuard, Config, NamedTempFile) {
-    use astria_core::generated::protocol::account::v1alpha1::NonceResponse;
+/// Start a mock sequencer server.
+async fn setup() -> (MockServer, Config, NamedTempFile) {
     Lazy::force(&TELEMETRY);
     let server = MockServer::start().await;
-    let startup_guard = mount_nonce_query_mock(
-        &server,
-        "accounts/nonce",
-        NonceResponse {
-            height: 0,
-            nonce: 0,
-        },
-    )
-    .await;
 
     let keyfile = NamedTempFile::new().unwrap();
     (&keyfile)
@@ -102,7 +104,7 @@ async fn setup() -> (MockServer, MockGuard, Config, NamedTempFile) {
         pretty_print: true,
         grpc_addr: "127.0.0.1:0".parse().unwrap(),
     };
-    (server, startup_guard, cfg, keyfile)
+    (server, cfg, keyfile)
 }
 
 /// Mount a mock for the `abci_query` endpoint.
@@ -181,6 +183,53 @@ async fn mount_broadcast_tx_sync_seq_actions_mock(server: &MockServer) -> MockGu
         .await
 }
 
+/// Mounts a `CometBFT` status response with a specified mock sequencer chain id
+async fn mount_genesis(server: &MockServer, mock_sequencer_chain_id: &str) {
+    Mock::given(body_partial_json(
+        json!({"jsonrpc": "2.0", "method": "genesis", "params": null}),
+    ))
+    .respond_with(ResponseTemplate::new(200).set_body_json(
+        tendermint_rpc::response::Wrapper::new_with_id(
+            tendermint_rpc::Id::uuid_v4(),
+            Some(
+                tendermint_rpc::endpoint::genesis::Response::<serde_json::Value> {
+                    genesis: Genesis {
+                        genesis_time: Time::from_unix_timestamp(1, 1).unwrap(),
+                        chain_id: mock_sequencer_chain_id.try_into().unwrap(),
+                        initial_height: 1,
+                        consensus_params: Params {
+                            block: tendermint::block::Size {
+                                max_bytes: 1024,
+                                max_gas: 1024,
+                                time_iota_ms: 1000,
+                            },
+                            evidence: tendermint::evidence::Params {
+                                max_age_num_blocks: 1000,
+                                max_age_duration: tendermint::evidence::Duration(
+                                    Duration::from_secs(3600),
+                                ),
+                                max_bytes: 1_048_576,
+                            },
+                            validator: ValidatorParams {
+                                pub_key_types: vec![tendermint::public_key::Algorithm::Ed25519],
+                            },
+                            version: None,
+                            abci: AbciParams::default(),
+                        },
+                        validators: vec![],
+                        app_hash: tendermint::hash::AppHash::default(),
+                        app_state: serde_json::Value::Null,
+                    },
+                },
+            ),
+            None,
+        ),
+    ))
+    .expect(1..)
+    .mount(server)
+    .await;
+}
+
 /// Helper to wait for the executor to connect to the mock sequencer
 async fn wait_for_startup(
     mut status: watch::Receiver<executor::Status>,
@@ -201,13 +250,15 @@ async fn wait_for_startup(
 
     Ok(())
 }
+
 /// Test to check that the executor sends a signed transaction to the sequencer as soon as it
 /// receives a `SequenceAction` that fills it beyond its `max_bundle_size`.
 #[tokio::test]
 async fn full_bundle() {
     // set up the executor, channel for writing seq actions, and the sequencer mock
-    let (sequencer, nonce_guard, cfg, _keyfile) = setup().await;
+    let (sequencer, cfg, _keyfile) = setup().await;
     let shutdown_token = CancellationToken::new();
+    mount_genesis(&sequencer, &cfg.sequencer_chain_id).await;
     let (executor, executor_handle) = executor::Builder {
         sequencer_url: cfg.sequencer_url.clone(),
         sequencer_chain_id: cfg.sequencer_chain_id.clone(),
@@ -220,8 +271,18 @@ async fn full_bundle() {
     .build()
     .unwrap();
 
+    let nonce_guard = mount_nonce_query_mock(
+        &sequencer,
+        "accounts/nonce",
+        NonceResponse {
+            height: 0,
+            nonce: 0,
+        },
+    )
+    .await;
     let status = executor.subscribe();
 
+    // executor.check_chain_ids().await.unwrap();
     let _executor_task = tokio::spawn(executor.run_until_stopped());
     // wait for sequencer to get the initial nonce request from sequencer
     wait_for_startup(status, nonce_guard).await.unwrap();
@@ -297,8 +358,9 @@ async fn full_bundle() {
 #[tokio::test]
 async fn bundle_triggered_by_block_timer() {
     // set up the executor, channel for writing seq actions, and the sequencer mock
-    let (sequencer, nonce_guard, cfg, _keyfile) = setup().await;
+    let (sequencer, cfg, _keyfile) = setup().await;
     let shutdown_token = CancellationToken::new();
+    mount_genesis(&sequencer, &cfg.sequencer_chain_id).await;
     let (executor, executor_handle) = executor::Builder {
         sequencer_url: cfg.sequencer_url.clone(),
         sequencer_chain_id: cfg.sequencer_chain_id.clone(),
@@ -311,8 +373,18 @@ async fn bundle_triggered_by_block_timer() {
     .build()
     .unwrap();
 
+    let nonce_guard = mount_nonce_query_mock(
+        &sequencer,
+        "accounts/nonce",
+        NonceResponse {
+            height: 0,
+            nonce: 0,
+        },
+    )
+    .await;
     let status = executor.subscribe();
 
+    // executor.check_chain_ids().await.unwrap();
     let _executor_task = tokio::spawn(executor.run_until_stopped());
 
     // wait for sequencer to get the initial nonce request from sequencer
@@ -381,8 +453,9 @@ async fn bundle_triggered_by_block_timer() {
 #[tokio::test]
 async fn two_seq_actions_single_bundle() {
     // set up the executor, channel for writing seq actions, and the sequencer mock
-    let (sequencer, nonce_guard, cfg, _keyfile) = setup().await;
+    let (sequencer, cfg, _keyfile) = setup().await;
     let shutdown_token = CancellationToken::new();
+    mount_genesis(&sequencer, &cfg.sequencer_chain_id).await;
     let (executor, executor_handle) = executor::Builder {
         sequencer_url: cfg.sequencer_url.clone(),
         sequencer_chain_id: cfg.sequencer_chain_id.clone(),
@@ -395,8 +468,18 @@ async fn two_seq_actions_single_bundle() {
     .build()
     .unwrap();
 
+    let nonce_guard = mount_nonce_query_mock(
+        &sequencer,
+        "accounts/nonce",
+        NonceResponse {
+            height: 0,
+            nonce: 0,
+        },
+    )
+    .await;
     let status = executor.subscribe();
 
+    // executor.check_chain_ids().await.unwrap();
     let _executor_task = tokio::spawn(executor.run_until_stopped());
 
     // wait for sequencer to get the initial nonce request from sequencer
@@ -469,3 +552,31 @@ async fn two_seq_actions_single_bundle() {
         );
     }
 }
+
+/// Test to check that executor's configured sequencer chain id and sequencer's actual chain id
+/// match
+#[tokio::test]
+async fn should_exit_if_mismatch_sequencer_chain_id() {
+    // set up sequencer mock
+    let (sequencer, cfg, _keyfile) = setup().await;
+    let shutdown_token = CancellationToken::new();
+
+    // mount a status response with an incorrect chain_id
+    mount_genesis(&sequencer, "bad-chain-id").await;
+
+    // build the executor with the correct chain_id
+    let (executor, _executor_handle) = executor::Builder {
+        sequencer_url: cfg.sequencer_url.clone(),
+        sequencer_chain_id: cfg.sequencer_chain_id.clone(),
+        private_key_file: cfg.private_key_file.clone(),
+        block_time_ms: cfg.block_time_ms,
+        max_bytes_per_bundle: cfg.max_bytes_per_bundle,
+        bundle_queue_capacity: cfg.bundle_queue_capacity,
+        shutdown_token: shutdown_token.clone(),
+    }
+    .build()
+    .unwrap();
+
+    // verify that the executor chain_id check results in an error
+    assert!(executor.run_until_stopped().await.is_err());
+}