test(starknet_integration_tests): test proposer-validator setup in fl…

…ow test

crates/starknet_integration_tests/src/flow_test_setup.rs

@@ -10,6 +10,7 @@ use starknet_consensus_manager::config::ConsensusManagerConfig;
 use starknet_gateway_types::errors::GatewaySpecError;
 use starknet_http_server::config::HttpServerConfig;
 use starknet_http_server::test_utils::HttpTestClient;
+use starknet_sequencer_node::config::node_config::SequencerNodeConfig;
 use starknet_sequencer_node::servers::run_component_servers;
 use starknet_sequencer_node::utils::create_node_modules;
 use starknet_task_executor::tokio_executor::TokioExecutor;
@@ -25,13 +26,15 @@ use crate::utils::{
     create_consensus_manager_configs_and_channels,
 };
 
-const PROPOSER_ID: usize = 0;
-const SEQUENCER_IDS: [usize; 1] = [PROPOSER_ID];
+const SEQUENCER_0: usize = 0;
+const SEQUENCER_1: usize = 1;
+const SEQUENCER_INDICES: [usize; 2] = [SEQUENCER_0, SEQUENCER_1];
 
 pub struct FlowTestSetup {
     // TODO(Tsabary): Remove this field.
     pub task_executor: TokioExecutor,
-    pub proposer: SequencerSetup,
+    pub sequencer_0: SequencerSetup,
+    pub sequencer_1: SequencerSetup,
 
     // Channels for consensus proposals, used for asserting the right transactions are proposed.
     pub consensus_proposals_channels: BroadcastTopicChannels<StreamMessage<ProposalPart>>,
@@ -46,28 +49,35 @@ impl FlowTestSetup {
 
         let accounts = tx_generator.accounts();
         let (mut consensus_manager_configs, consensus_proposals_channels) =
-            create_consensus_manager_configs_and_channels(SEQUENCER_IDS.len());
+            create_consensus_manager_configs_and_channels(SEQUENCER_INDICES.len());
 
-        // Take the first config for every sequencer node.
-        let proposer_consensus_manager_config = consensus_manager_configs.remove(0);
-        let proposer = SequencerSetup::new(
+        // Take the first config for every sequencer node, and create nodes one after the other in
+        // order to make sure the ports are not overlapping.
+        let sequencer_0_consensus_manager_config = consensus_manager_configs.remove(0);
+        let sequencer_0 = SequencerSetup::new(
             accounts.clone(),
-            PROPOSER_ID,
+            SEQUENCER_0,
             chain_info.clone(),
             &task_executor,
-            proposer_consensus_manager_config,
+            sequencer_0_consensus_manager_config,
         )
         .await;
 
-        Self { task_executor, proposer, consensus_proposals_channels }
-    }
+        let sequencer_1_consensus_manager_config = consensus_manager_configs.remove(0);
+        let sequencer_1 = SequencerSetup::new(
+            accounts,
+            SEQUENCER_1,
+            chain_info,
+            &task_executor,
+            sequencer_1_consensus_manager_config,
+        )
+        .await;
 
-    pub async fn assert_add_tx_success(&self, tx: RpcTransaction) -> TransactionHash {
-        self.proposer.add_tx_http_client.assert_add_tx_success(tx).await
+        Self { task_executor, sequencer_0, sequencer_1, consensus_proposals_channels }
     }
 
     pub async fn assert_add_tx_error(&self, tx: RpcTransaction) -> GatewaySpecError {
-        self.proposer.add_tx_http_client.assert_add_tx_error(tx).await
+        self.sequencer_0.add_tx_http_client.assert_add_tx_error(tx).await
     }
 }
 
@@ -84,6 +94,8 @@ pub struct SequencerSetup {
 
     // Handle of the sequencer node.
     pub sequencer_node_handle: JoinHandle<Result<(), anyhow::Error>>,
+
+    pub config: SequencerNodeConfig,
 }
 
 impl SequencerSetup {
@@ -139,6 +151,11 @@ impl SequencerSetup {
             batcher_storage_file_handle: storage_for_test.batcher_storage_handle,
             rpc_storage_file_handle: storage_for_test.rpc_storage_handle,
             sequencer_node_handle,
+            config,
         }
     }
+
+    pub async fn assert_add_tx_success(&self, tx: RpcTransaction) -> TransactionHash {
+        self.add_tx_http_client.assert_add_tx_success(tx).await
+    }
 }

crates/starknet_integration_tests/tests/end_to_end_flow_test.rs

@@ -2,7 +2,7 @@ use std::collections::HashSet;
 
 use futures::StreamExt;
 use mempool_test_utils::starknet_api_test_utils::MultiAccountTransactionGenerator;
-use papyrus_consensus::types::{ValidatorId, DEFAULT_VALIDATOR_ID};
+use papyrus_consensus::types::ValidatorId;
 use papyrus_network::network_manager::BroadcastTopicChannels;
 use papyrus_protobuf::consensus::{
     ProposalFin,
@@ -47,20 +47,26 @@ async fn end_to_end_flow(mut tx_generator: MultiAccountTransactionGenerator) {
     let heights_to_build = next_height.iter_up_to(LAST_HEIGHT.unchecked_next());
     let expected_content_ids = [
         Felt::from_hex_unchecked(
-            "0x7d62e32fd8f1a12104a5d215af26ec0f362da81af3d14c24e08e46976cdfbf5",
+            "0x457e9172b9c70fb4363bb3ff31bf778d8f83828184a9a3f9badadc497f2b954",
         ),
         Felt::from_hex_unchecked(
-            "0x259aeaad847bffe6c342998c4510e5e474577219cfbb118f5cb2f2286260d52",
+            "0x572373fe992ac8c2413d5e727036316023ed6a2e8a2256b4952e223969e0221",
         ),
     ];
 
-    // Buld multiple heights to ensure heights are committed.
+    let sequencers = [&mock_running_system.sequencer_0, &mock_running_system.sequencer_1];
+    let mut expected_proposer_iter = sequencers.iter().cycle();
+    // We start at height 1, so we need to skip the proposer of the initial height.
+    expected_proposer_iter.next().unwrap();
+
+    // Build multiple heights to ensure heights are committed.
     for (height, expected_content_id) in itertools::zip_eq(heights_to_build, expected_content_ids) {
         debug!("Starting height {}.", height);
+        let expected_proposer = expected_proposer_iter.next().unwrap();
         // Create and send transactions.
         let expected_batched_tx_hashes =
             run_integration_test_scenario(&mut tx_generator, &mut |tx| {
-                mock_running_system.assert_add_tx_success(tx)
+                expected_proposer.assert_add_tx_success(tx)
             })
             .await;
         // TODO(Dan, Itay): Consider adding a utility function that waits for something to happen.
@@ -71,6 +77,7 @@ async fn end_to_end_flow(mut tx_generator: MultiAccountTransactionGenerator) {
                 &expected_batched_tx_hashes,
                 height,
                 expected_content_id,
+                expected_proposer.config.consensus_manager_config.consensus_config.validator_id,
             ),
         )
         .await
@@ -83,6 +90,7 @@ async fn listen_to_broadcasted_messages(
     expected_batched_tx_hashes: &[TransactionHash],
     expected_height: BlockNumber,
     expected_content_id: Felt,
+    expected_proposer_id: ValidatorId,
 ) {
     let chain_id = CHAIN_ID_FOR_TESTS.clone();
     let broadcasted_messages_receiver =
@@ -92,7 +100,7 @@ async fn listen_to_broadcasted_messages(
         height: expected_height,
         round: 0,
         valid_round: None,
-        proposer: ValidatorId::from(DEFAULT_VALIDATOR_ID),
+        proposer: expected_proposer_id,
     };
     let expected_proposal_fin = ProposalFin { proposal_content_id: BlockHash(expected_content_id) };