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/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_gateway/src/gateway.rs

@@ -11,7 +11,7 @@ use starknet_mempool_types::communication::{AddTransactionArgsWrapper, SharedMem
 use starknet_mempool_types::mempool_types::{AccountState, AddTransactionArgs};
 use starknet_sequencer_infra::component_definitions::ComponentStarter;
 use starknet_sierra_compile::config::SierraToCasmCompilationConfig;
-use tracing::{error, instrument, Span};
+use tracing::{error, info, instrument, Span};
 
 use crate::compilation::GatewayCompiler;
 use crate::config::{GatewayConfig, RpcStateReaderConfig};
@@ -64,6 +64,7 @@ impl Gateway {
         tx: RpcTransaction,
         p2p_message_metadata: Option<BroadcastedMessageMetadata>,
     ) -> GatewayResult<TransactionHash> {
+        info!("Processing tx: {:?}", tx);
         let blocking_task = ProcessTxBlockingTask::new(self, tx);
         // Run the blocking task in the current span.
         let curr_span = Span::current();

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::node_runner::{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,20 +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.
-    integration_test_setup
-        .is_alive_test_client
-        .await_alive(5000, 50)
-        .await
-        .expect("Node should be 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;
@@ -83,19 +92,26 @@ 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");
 
-    await_block(5000, EXPECTED_BLOCK_NUMBER, 15, &batcher_storage_reader)
+    await_block(5000, EXPECTED_BLOCK_NUMBER, 30, &batcher_storage_reader)
         .await
         .expect("Block number should have been reached.");
 
-    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

@@ -33,8 +33,8 @@ const SEQUENCER_1: usize = 1;
 const SEQUENCER_INDICES: [usize; 2] = [SEQUENCER_0, SEQUENCER_1];
 
 pub struct FlowTestSetup {
-    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>>,
@@ -57,15 +57,16 @@ 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(),
             sequencer_0_consensus_manager_config,
             sequencer_0_mempool_p2p_config,
         )
         .await;
-        let sequencer_1 = SequencerSetup::new(
+
+        let sequencer_1 = FlowSequencerSetup::new(
             accounts,
             SEQUENCER_1,
             chain_info,
@@ -82,7 +83,7 @@ impl FlowTestSetup {
     }
 }
 
-pub struct SequencerSetup {
+pub struct FlowSequencerSetup {
     /// Used to differentiate between different sequencer nodes.
     pub sequencer_index: usize,
 
@@ -103,7 +104,7 @@ pub struct SequencerSetup {
     pub is_alive_test_client: IsAliveClient,
 }
 
-impl SequencerSetup {
+impl FlowSequencerSetup {
     #[instrument(skip(accounts, chain_info, consensus_manager_config), level = "debug")]
     pub async fn new(
         accounts: Vec<Contract>,