test(starknet_integration_tests): add second node

crates/papyrus_config/src/dumping.rs

@@ -363,7 +363,7 @@ pub fn ser_pointer_target_required_param(
 /// Updates entries in the map to point to these targets, replacing values of entries that match
 /// the target parameter paths to contain only the name of the target they point to.
 /// Fails if a param is not pointing to a same-named pointer target nor whitelisted.
-pub(crate) fn combine_config_map_and_pointers(
+pub fn combine_config_map_and_pointers(
     mut config_map: BTreeMap<ParamPath, SerializedParam>,
     pointers: &ConfigPointers,
     non_pointer_params: &Pointers,

crates/papyrus_network/src/gossipsub_impl.rs

@@ -6,10 +6,10 @@ use crate::mixed_behaviour;
 use crate::mixed_behaviour::BridgedBehaviour;
 use crate::sqmr::Bytes;
 
-#[cfg(test)]
+// #[cfg(test)]
 pub type Topic = gossipsub::IdentTopic;
-#[cfg(not(test))]
-pub type Topic = gossipsub::Sha256Topic;
+// #[cfg(not(test))]
+// pub type Topic = gossipsub::Sha256Topic;
 
 #[derive(Debug)]
 pub enum ExternalEvent {

crates/papyrus_network/src/network_manager/test_utils.rs

@@ -188,7 +188,7 @@ where
     const BUFFER_SIZE: usize = 1000;
 
     let mut channels_configs = create_connected_network_configs(n_configs + 1);
-    let broadcast_channels = channels_configs.remove(0);
+    let broadcast_channels = channels_configs.pop().unwrap();
 
     let mut channels_network_manager = NetworkManager::new(broadcast_channels, None);
     let broadcast_channels =

crates/starknet_integration_tests/Cargo.toml

@@ -18,6 +18,7 @@ indexmap.workspace = true
 infra_utils.workspace = true
 mempool_test_utils.workspace = true
 papyrus_common.workspace = true
+papyrus_config.workspace = true
 papyrus_consensus.workspace = true
 papyrus_execution.workspace = true
 papyrus_network = { workspace = true, features = ["testing"] }

crates/starknet_integration_tests/src/config_utils.rs

@@ -3,7 +3,8 @@ use std::io::Write;
 use std::net::SocketAddr;
 use std::path::PathBuf;
 
-use serde_json::{json, Value};
+use papyrus_config::dumping::{combine_config_map_and_pointers, SerializeConfig};
+use serde_json::{json, Map, Value};
 use starknet_sequencer_infra::component_definitions::{
     LocalServerConfig,
     RemoteClientConfig,
@@ -16,7 +17,11 @@ use starknet_sequencer_node::config::component_execution_config::{
     ReactiveComponentExecutionConfig,
     ReactiveComponentExecutionMode,
 };
-use starknet_sequencer_node::config::node_config::SequencerNodeConfig;
+use starknet_sequencer_node::config::node_config::{
+    SequencerNodeConfig,
+    CONFIG_NON_POINTERS_WHITELIST,
+    CONFIG_POINTERS,
+};
 use starknet_sequencer_node::config::test_utils::RequiredParams;
 use tracing::info;
 // TODO(Tsabary): Move here all config-related functions from "integration_test_utils.rs".
@@ -55,20 +60,29 @@ pub(crate) fn dump_config_file_changes(
 ) -> PathBuf {
     // Dump config changes file for the sequencer node.
     // TODO(Tsabary): auto dump the entirety of RequiredParams fields.
-    let json_data = config_fields_to_json!(
+    let required_params_json = config_fields_to_json!(
         required_params.chain_id,
         required_params.eth_fee_token_address,
         required_params.strk_fee_token_address,
         required_params.validator_id,
-        config.rpc_state_reader_config.url,
-        config.batcher_config.storage.db_config.path_prefix,
-        config.http_server_config.ip,
-        config.http_server_config.port,
-        config.consensus_manager_config.consensus_config.start_height,
-        config.state_sync_config.storage_config.db_config.path_prefix,
-        config.state_sync_config.network_config.tcp_port,
     );
-    let node_config_path = dump_json_data(json_data, NODE_CONFIG_CHANGES_FILE_PATH, dir);
+
+    // Create the entire mapping of the config and the pointers, without the required params.
+    let config_as_map = combine_config_map_and_pointers(
+        config.dump(),
+        &CONFIG_POINTERS,
+        &CONFIG_NON_POINTERS_WHITELIST,
+    )
+    .unwrap();
+
+    // Extract only the required fields from the config map.
+    let mut preset = config_to_preset(&config_as_map, "value");
+
+    // Add the required params to the preset.
+    add_required_params_to_preset(&mut preset, &required_params_json);
+
+    // Dump the preset to a file, return its path.
+    let node_config_path = dump_json_data(preset, NODE_CONFIG_CHANGES_FILE_PATH, dir);
     assert!(node_config_path.exists(), "File does not exist: {:?}", node_config_path);
 
     node_config_path
@@ -157,3 +171,36 @@ pub async fn get_remote_flow_test_config() -> Vec<ComponentConfig> {
         get_non_http_component_config(gateway_socket).await,
     ]
 }
+
+fn config_to_preset(config_map: &Value, inner_key: &str) -> Value {
+    // Ensure the config_map is a JSON object
+    if let Value::Object(map) = config_map {
+        let mut result = Map::new();
+
+        for (key, value) in map {
+            if let Value::Object(inner_map) = value {
+                // Extract the value for the specified inner_key
+                if let Some(inner_value) = inner_map.get(inner_key) {
+                    // Add it to the result map
+                    result.insert(key.clone(), inner_value.clone());
+                }
+            }
+        }
+
+        // Return the transformed result as a JSON object
+        Value::Object(result)
+    } else {
+        // If the input is not an object, return an empty object
+        Value::Object(Map::new())
+    }
+}
+
+fn add_required_params_to_preset(preset: &mut Value, required_params: &Value) {
+    if let (Value::Object(preset_map), Value::Object(required_params_map)) =
+        (preset, required_params)
+    {
+        for (key, value) in required_params_map {
+            preset_map.insert(key.clone(), value.clone());
+        }
+    }
+}

crates/starknet_integration_tests/src/end_to_end_integration.rs

@@ -1,4 +1,5 @@
 use infra_utils::run_until::run_until;
+use infra_utils::tracing::{CustomLogger, TraceLevel};
 use mempool_test_utils::starknet_api_test_utils::{AccountId, MultiAccountTransactionGenerator};
 use papyrus_execution::execution_utils::get_nonce_at;
 use papyrus_storage::state::StateStorageReader;
@@ -8,6 +9,7 @@ use starknet_api::core::{ContractAddress, Nonce};
 use starknet_api::state::StateNumber;
 use starknet_sequencer_node::test_utils::compilation::{get_node_executable_path, spawn_run_node};
 use starknet_types_core::felt::Felt;
+use tokio::join;
 use tracing::info;
 
 use crate::integration_test_setup::IntegrationTestSetup;
@@ -43,7 +45,12 @@ async fn await_block(
     let condition = |&latest_block_number: &BlockNumber| latest_block_number >= target_block_number;
     let get_latest_block_number_closure = || async move { get_latest_block_number(storage_reader) };
 
-    run_until(interval, max_attempts, get_latest_block_number_closure, condition, None)
+    let logger = CustomLogger::new(
+        TraceLevel::Info,
+        Some("Waiting for storage to include block".to_string()),
+    );
+
+    run_until(interval, max_attempts, get_latest_block_number_closure, condition, Some(logger))
         .await
         .ok_or(())
 }
@@ -58,19 +65,22 @@ pub async fn end_to_end_integration(mut tx_generator: MultiAccountTransactionGen
     // Creating the storage for the test.
     let integration_test_setup = IntegrationTestSetup::new_from_tx_generator(&tx_generator).await;
 
-    info!("Running sequencer node.");
-    let node_run_handle = spawn_run_node(integration_test_setup.node_config_path).await;
+    let node_0_run_handle =
+        spawn_run_node(integration_test_setup.sequencer_0.node_config_path.clone());
+    let node_1_run_handle =
+        spawn_run_node(integration_test_setup.sequencer_1.node_config_path.clone());
 
-    // Wait for the node to start.
-    match integration_test_setup.is_alive_test_client.await_alive(5000, 50).await {
-        Ok(_) => {}
-        Err(_) => panic!("Node is not alive."),
-    }
+    info!("Running sequencers.");
+    let (node_0_run_task, node_1_run_task) = join!(node_0_run_handle, node_1_run_handle);
+
+    // Wait for the nodes to start.
+    integration_test_setup.await_alive(5000, 50).await;
 
     info!("Running integration test simulator.");
 
-    let send_rpc_tx_fn =
-        &mut |rpc_tx| integration_test_setup.add_tx_http_client.assert_add_tx_success(rpc_tx);
+    let send_rpc_tx_fn = &mut |rpc_tx| {
+        integration_test_setup.sequencer_0.add_tx_http_client.assert_add_tx_success(rpc_tx)
+    };
 
     const ACCOUNT_ID_0: AccountId = 0;
     let n_txs = 50;
@@ -81,20 +91,27 @@ pub async fn end_to_end_integration(mut tx_generator: MultiAccountTransactionGen
     info!("Awaiting until {EXPECTED_BLOCK_NUMBER} blocks have been created.");
 
     let (batcher_storage_reader, _) =
-        papyrus_storage::open_storage(integration_test_setup.batcher_storage_config)
+        papyrus_storage::open_storage(integration_test_setup.sequencer_0.batcher_storage_config)
             .expect("Failed to open batcher's storage");
 
-    match await_block(5000, EXPECTED_BLOCK_NUMBER, 15, &batcher_storage_reader).await {
+    match await_block(5000, EXPECTED_BLOCK_NUMBER, 30, &batcher_storage_reader).await {
         Ok(_) => {}
         Err(_) => panic!("Did not reach expected block number."),
     }
 
-    info!("Shutting the node down.");
-    node_run_handle.abort();
-    let res = node_run_handle.await;
+    info!("Shutting node 0 down.");
+    node_0_run_task.abort();
+    let res = node_0_run_task.await;
+    assert!(
+        res.expect_err("Node 0 should have been stopped.").is_cancelled(),
+        "Node 0 should have been stopped."
+    );
+    info!("Shutting node 1 down.");
+    node_1_run_task.abort();
+    let res = node_1_run_task.await;
     assert!(
-        res.expect_err("Node should have been stopped.").is_cancelled(),
-        "Node should have been stopped."
+        res.expect_err("Node 1 should have been stopped.").is_cancelled(),
+        "Node 1 should have been stopped."
     );
 
     info!("Verifying tx sender account nonce.");

crates/starknet_integration_tests/src/flow_test_setup.rs

@@ -35,8 +35,8 @@ const SEQUENCER_INDICES: [usize; 2] = [SEQUENCER_0, SEQUENCER_1];
 pub struct FlowTestSetup {
     // TODO(Tsabary): Remove this field.
     pub task_executor: TokioExecutor,
-    pub sequencer_0: SequencerSetup,
-    pub sequencer_1: SequencerSetup,
+    pub sequencer_0: FlowSequencerSetup,
+    pub sequencer_1: FlowSequencerSetup,
 
     // Channels for consensus proposals, used for asserting the right transactions are proposed.
     pub consensus_proposals_channels: BroadcastTopicChannels<StreamMessage<ProposalPart>>,
@@ -61,7 +61,7 @@ impl FlowTestSetup {
             2] = mempool_p2p_configs.try_into().unwrap();
 
         // Create nodes one after the other in order to make sure the ports are not overlapping.
-        let sequencer_0 = SequencerSetup::new(
+        let sequencer_0 = FlowSequencerSetup::new(
             accounts.clone(),
             SEQUENCER_0,
             chain_info.clone(),
@@ -70,7 +70,8 @@ impl FlowTestSetup {
             sequencer_0_mempool_p2p_config,
         )
         .await;
-        let sequencer_1 = SequencerSetup::new(
+
+        let sequencer_1 = FlowSequencerSetup::new(
             accounts,
             SEQUENCER_1,
             chain_info,
@@ -88,7 +89,7 @@ impl FlowTestSetup {
     }
 }
 
-pub struct SequencerSetup {
+pub struct FlowSequencerSetup {
     /// Used to differentiate between different sequencer nodes.
     pub sequencer_index: usize,
 
@@ -106,7 +107,7 @@ pub struct SequencerSetup {
     pub config: SequencerNodeConfig,
 }
 
-impl SequencerSetup {
+impl FlowSequencerSetup {
     #[instrument(
         skip(accounts, chain_info, task_executor, consensus_manager_config),
         level = "debug"