conductor: replace gossipnet by a subscription to cometbft/sequencer (#…

…383) ## Summary This patch replaces gossipnet by a websocket stream of new blocks directly from sequencer. ## Background Adding a p2p layer adds a lot of complexity to astria services. With this patch conductor reads new blocks directly off of cometbft/sequencer, making it much simpler. For background reading, see this issue: #325 ## Changes - Replace gossipnet directly by a stream of new blocks using the cometbft subscrition RPC. - No other changes in regard to how blocks are further processed has been made. ## Testing We are relying on the rust tendermint-rpc crate functioning properly. As a follow-up it makes sense to create blackbox tests for conductor, which are currently absent. ## Breaking Changelist - Reading over p2p is no longer supported. Deployments must be updated and wired to read off of cometbft/sequenceer. ## Related Issues Precursor PRs unblocking this work: #375 #376 #382 closes #332 --------- Co-authored-by: Emilia Hane <emiliaha95@gmail.com>
astriaorg · Sep 20, 2023 · 9d75fb5 · 9d75fb5
1 parent 1df160e
commit 9d75fb5
Show file tree

Hide file tree

Showing 11 changed files with 120 additions and 270 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/crates/astria-conductor/Cargo.toml b/crates/astria-conductor/Cargo.toml
@@ -23,15 +23,14 @@ serde_json = { workspace = true }
 sha2 = { workspace = true }
 tendermint = { workspace = true }
 tendermint-proto = { workspace = true }
-tendermint-rpc = { workspace = true }
 tokio = { workspace = true, features = ["macros", "rt-multi-thread", "signal"] }
 tonic = { workspace = true }
 tracing = { workspace = true }
 
 astria-proto = { path = "../astria-proto", features = ["client"] }
-astria-gossipnet = { path = "../astria-gossipnet" }
-astria-sequencer-client = { path = "../astria-sequencer-client", features = [
+sequencer-client = { package = "astria-sequencer-client", path = "../astria-sequencer-client", features = [
   "http",
+  "websocket",
 ] }
 astria-sequencer-relayer = { path = "../astria-sequencer-relayer" }
 astria-sequencer-types = { path = "../astria-sequencer-types" }

diff --git a/crates/astria-conductor/README.md b/crates/astria-conductor/README.md
@@ -32,33 +32,6 @@ load it and run locally:
 just run
 ```
 
-### Additional env variables
-
-#### Bootnodes
-
-You can also connect directly to a node - just add a bootnode address to the .env file
-
-```bash
-ASTRIA_CONDUCTOR_BOOTNODES="/ip4/127.0.0.1/tcp/34471/p2p/12D3KooWDCHwgGetpJuHknJqv2dNbYpe3LqgH8BKrsYHV9ALpAj8"
-```
-
-#### libp2p options
-
-You can add a libp2p private key or port
-
-```bash
-ASTRIA_CONDUCTOR_LIBP2P_PRIVATE_KEY="{{your key}}"
-ASTRIA_CONDUCTOR_LIBP2P_PORT="{{your port}}"
-```
-
-#### Celestia JWT bearer token
-
-You can add a JWT token that's used in celestia jsonrpc calls
-
-```bash
-ASTRIA_CONDUCTOR_CELESTIA_BEARER_TOKEN="{{your token}}"
-```
-
 ### Running tests
 
 ```bash

diff --git a/crates/astria-conductor/local.env.example b/crates/astria-conductor/local.env.example
@@ -7,31 +7,24 @@ ASTRIA_CONDUCTOR_LOG="astria_conductor=info"
 ASTRIA_CONDUCTOR_CELESTIA_BEARER_TOKEN="<JWT Bearer token>"
 
 # Data Availability service url (Celestia node in this case)
-# This url is used to read finalized astria blocks from the Data Availability layer
+# This url is used to read astria blocks from the Data Availability layer
 ASTRIA_CONDUCTOR_CELESTIA_NODE_URL="http://127.0.0.1:26659"
 
 # Tendermint URL
-# This url is used to talk to the astria sequener to eagerly read blocks before
-# they are finalized and available from the data avaialability layer
+# This url is used to talk to the astria sequencer to get validator sets
 ASTRIA_CONDUCTOR_TENDERMINT_URL="http://127.0.0.1:26657"
 
 # The chain id of the chain that is being read from the astria-sequencer or the
 # Data Availability layer
 ASTRIA_CONDUCTOR_CHAIN_ID="ethereum"
 
-# A list of bootnodes that conductor will use to join the gossipnet p2p network.
-# Optional: if left empty conductor will be a single node in the gossipnet,
-# waiting for peers to join it.
-ASTRIA_CONDUCTOR_BOOTNODES=
-
 # Execution RPC URL
 ASTRIA_CONDUCTOR_EXECUTION_RPC_URL="http://127.0.0.1:50051"
 
 # disable block finalization
 ASTRIA_CONDUCTOR_DISABLE_FINALIZATION=false
 
-# The path to the libp2p private key.
-ASTRIA_CONDUCTOR_LIBP2P_PRIVATE_KEY=/home/user/.gossipnet/libp2p.priv
-
-# libp2p Port
-ASTRIA_CONDUCTOR_LIBP2P_PORT=2451
+# The URL to a fully trusted CometBFT/Sequencer to subscribe to new blocks
+# over websocket using cometbft's RPC.
+# 127.0.0.1:26657 is the default socket address in comebft's `rpc.laddr` setting.
+ASTRIA_CONDUCTOR_SEQUENCER_URL="ws://127.0.0.1:26657"
diff --git a/crates/astria-conductor/src/block_verifier.rs b/crates/astria-conductor/src/block_verifier.rs
@@ -5,11 +5,7 @@ use astria_sequencer_relayer::data_availability::{
     SequencerNamespaceData,
     SignedNamespaceData,
 };
-use astria_sequencer_types::{
-    calculate_last_commit_hash,
-    RawSequencerBlockData,
-    SequencerBlockData,
-};
+use astria_sequencer_types::calculate_last_commit_hash;
 use color_eyre::eyre::{
     self,
     bail,
@@ -21,6 +17,11 @@ use ed25519_consensus::{
     VerificationKey,
 };
 use prost::Message;
+use sequencer_client::{
+    tendermint::endpoint::validators,
+    Client as _,
+    HttpClient,
+};
 use tendermint::{
     account::Id as AccountId,
     block::Id as BlockId,
@@ -31,11 +32,6 @@ use tendermint::{
         CanonicalVote,
     },
 };
-use tendermint_rpc::{
-    endpoint::validators::Response as ValidatorSet,
-    Client,
-    HttpClient,
-};
 use tracing::instrument;
 
 /// `BlockVerifier` is responsible for verifying the correctness of a block
@@ -71,11 +67,11 @@ impl BlockVerifier {
         );
         let validator_set = self
             .sequencer_client
-            .validators(height, tendermint_rpc::Paging::Default)
+            .validators(height, sequencer_client::tendermint::Paging::Default)
             .await
             .wrap_err("failed to get validator set")?;
 
-        validate_signed_namespace_data(validator_set, data).await
+        validate_signed_namespace_data(validator_set, data)
     }
 
     /// validates `RollupNamespaceData` received from Celestia.
@@ -89,7 +85,7 @@ impl BlockVerifier {
     ) -> eyre::Result<()> {
         self.validate_sequencer_namespace_data(sequencer_namespace_data)
             .await
-            .context("failed to validate sequencer block header and last commit")?;
+            .wrap_err("failed to validate sequencer block header and last commit")?;
 
         // validate that rollup data was included in the sequencer block
         rollup_data
@@ -98,46 +94,6 @@ impl BlockVerifier {
         Ok(())
     }
 
-    /// validates [`SequencerBlockData`] received from the gossip network.
-    pub(crate) async fn validate_sequencer_block_data(
-        &self,
-        block: &SequencerBlockData,
-    ) -> eyre::Result<()> {
-        // TODO(https://github.com/astriaorg/astria/issues/309): remove this clone by not gossiping the entire [`SequencerBlockData`]
-        let RawSequencerBlockData {
-            block_hash,
-            header,
-            last_commit,
-            rollup_data,
-            action_tree_root,
-            action_tree_root_inclusion_proof,
-            chain_ids_commitment,
-        } = block.clone().into_raw();
-        let rollup_chain_ids = rollup_data
-            .into_keys()
-            .collect::<Vec<astria_sequencer_types::ChainId>>();
-        let data = SequencerNamespaceData {
-            block_hash,
-            header,
-            last_commit,
-            rollup_chain_ids,
-            action_tree_root,
-            action_tree_root_inclusion_proof,
-            chain_ids_commitment,
-        };
-
-        self.validate_sequencer_namespace_data(&data).await?;
-
-        // TODO(https://github.com/astriaorg/astria/issues/309): validate that the transactions in
-        // the block result in the correct data_hash
-        // however this requires updating [`SequencerBlockData`] to contain inclusion proofs for
-        // each of its rollup datas; we can instead update the gossip network to *not*
-        // gossip entire [`SequencerBlockData`] but instead gossip headers, while each
-        // conductor subscribes only to rollup data for its own namespace. then, we can
-        // simply use [`validate_rollup_data`] the same way we use it for data pulled from DA.
-        Ok(())
-    }
-
     /// performs various validation checks on the SequencerBlock received from either gossip or
     /// Celestia.
     ///
@@ -162,24 +118,24 @@ impl BlockVerifier {
         // get the validator set for this height
         let current_validator_set = self
             .sequencer_client
-            .validators(height, tendermint_rpc::Paging::Default)
+            .validators(height, sequencer_client::tendermint::Paging::Default)
             .await
             .wrap_err("failed to get validator set")?;
 
         // get validator set for the previous height, as the commit contained
         // in the block is for the previous height
         let parent_validator_set = self
             .sequencer_client
-            .validators(height - 1, tendermint_rpc::Paging::Default)
+            .validators(height - 1, sequencer_client::tendermint::Paging::Default)
             .await
             .wrap_err("failed to get validator set")?;
 
-        validate_sequencer_namespace_data(current_validator_set, parent_validator_set, data).await
+        validate_sequencer_namespace_data(current_validator_set, parent_validator_set, data)
     }
 }
 
-async fn validate_signed_namespace_data(
-    validator_set: ValidatorSet,
+fn validate_signed_namespace_data(
+    validator_set: validators::Response,
     data: &SignedNamespaceData<SequencerNamespaceData>,
 ) -> eyre::Result<()> {
     // verify the block signature
@@ -204,9 +160,9 @@ async fn validate_signed_namespace_data(
     Ok(())
 }
 
-async fn validate_sequencer_namespace_data(
-    current_validator_set: ValidatorSet,
-    parent_validator_set: ValidatorSet,
+fn validate_sequencer_namespace_data(
+    current_validator_set: validators::Response,
+    parent_validator_set: validators::Response,
     data: &SequencerNamespaceData,
 ) -> eyre::Result<()> {
     let SequencerNamespaceData {
@@ -250,11 +206,8 @@ async fn validate_sequencer_namespace_data(
             // verify that the validator votes on the previous block have >2/3 voting power
             let last_commit = last_commit.clone();
             let chain_id = header.chain_id.clone();
-            tokio::task::spawn_blocking(move || -> eyre::Result<()> {
-                ensure_commit_has_quorum(&last_commit, &parent_validator_set, chain_id.as_ref())
-            })
-            .await?
-            .wrap_err("failed to ensure commit has quorum")?
+            ensure_commit_has_quorum(&last_commit, &parent_validator_set, chain_id.as_ref())
+                .wrap_err("failed to ensure commit has quorum")?
 
             // TODO: commit is for previous block; how do we handle this? (#50)
         }
@@ -324,7 +277,7 @@ fn public_key_bytes_to_address(public_key: &tendermint::PublicKey) -> eyre::Resu
 #[instrument]
 fn ensure_commit_has_quorum(
     commit: &tendermint::block::Commit,
-    validator_set: &ValidatorSet,
+    validator_set: &validators::Response,
     chain_id: &str,
 ) -> eyre::Result<()> {
     if commit.height != validator_set.block_height {
@@ -468,7 +421,7 @@ fn verify_vote_signature(
 /// returns the proposer given the current set by ordering the validators by proposer priority.
 /// the validator with the highest proposer priority is the proposer.
 /// TODO: could there ever be two validators with the same priority?
-fn get_proposer(validator_set: &ValidatorSet) -> eyre::Result<Validator> {
+fn get_proposer(validator_set: &validators::Response) -> eyre::Result<Validator> {
     validator_set
         .validators
         .iter()
@@ -497,7 +450,7 @@ mod test {
 
     use super::*;
 
-    fn make_test_validator_set(height: u32) -> (ValidatorSet, account::Id) {
+    fn make_test_validator_set(height: u32) -> (validators::Response, account::Id) {
         use rand::rngs::OsRng;
 
         let signing_key = ed25519_consensus::SigningKey::new(OsRng);
@@ -516,13 +469,13 @@ mod test {
         };
 
         (
-            ValidatorSet::new(height.into(), vec![validator], 1),
+            validators::Response::new(height.into(), vec![validator], 1),
             address,
         )
     }
 
-    #[tokio::test]
-    async fn validate_sequencer_namespace_data_last_commit_none_ok() {
+    #[test]
+    fn validate_sequencer_namespace_data_last_commit_none_ok() {
         let action_tree = MerkleTree::from_leaves(vec![vec![1, 2, 3], vec![4, 5, 6]]);
         let action_tree_root = action_tree.root();
 
@@ -553,7 +506,6 @@ mod test {
             make_test_validator_set(height - 1).0,
             &sequencer_namespace_data,
         )
-        .await
         .unwrap();
     }
 
@@ -604,7 +556,6 @@ mod test {
             make_test_validator_set(height - 1).0,
             &sequencer_namespace_data,
         )
-        .await
         .unwrap();
         rollup_namespace_data
             .verify_inclusion_proof(sequencer_namespace_data.action_tree_root)
@@ -647,7 +598,8 @@ mod test {
         // curl http://localhost:26657/commit?height=79
         let validator_set_str = r#"{"block_height":"79","validators":[{"address":"D223B03AE01B4A0296053E01A41AE1E2F9CDEBC9","pub_key":{"type":"tendermint/PubKeyEd25519","value":"tyPnz5GGblrx3PBjQRxZOHbzsPEI1E8lOh62QoPSWLw="},"voting_power":"10","proposer_priority":"0"}],"count":"1","total":"1"}"#;
         let commit_str = r#"{"height":"79","round":0,"block_id":{"hash":"74BD4E7F7EF902A84D55589F2AA60B332F1C2F34DDE7652C80BFEB8E7471B1DA","parts":{"total":1,"hash":"7632FFB5D84C3A64279BC9EA86992418ED23832C66E0C3504B7025A9AF42C8C4"}},"signatures":[{"block_id_flag":2,"validator_address":"D223B03AE01B4A0296053E01A41AE1E2F9CDEBC9","timestamp":"2023-07-05T19:02:55.206600022Z","signature":"qy9vEjqSrF+8sD0K0IAXA398xN1s3QI2rBBDbBMWf0rw0L+B9Z92DZEptf6bPYWuKUFdEc0QFKhUMQA8HjBaAw=="}]}"#;
-        let validator_set = serde_json::from_str::<ValidatorSet>(validator_set_str).unwrap();
+        let validator_set =
+            serde_json::from_str::<validators::Response>(validator_set_str).unwrap();
         let commit = serde_json::from_str::<Commit>(commit_str).unwrap();
         ensure_commit_has_quorum(&commit, &validator_set, "test-chain-g3ejvw").unwrap();
     }
@@ -658,7 +610,7 @@ mod test {
             general_purpose::STANDARD,
             Engine as _,
         };
-        let validator_set = ValidatorSet::new(
+        let validator_set = validators::Response::new(
             79u32.into(),
             vec![Validator {
                 name: None,