diff --git a/crates/apollo_integration_tests/src/bin/sequencer_node_end_to_end_integration_tests/integration_test_central_and_p2p_sync_flow.rs b/crates/apollo_integration_tests/src/bin/sequencer_node_end_to_end_integration_tests/integration_test_central_and_p2p_sync_flow.rs index b2232144745..712e9c92c6a 100644 --- a/crates/apollo_integration_tests/src/bin/sequencer_node_end_to_end_integration_tests/integration_test_central_and_p2p_sync_flow.rs +++ b/crates/apollo_integration_tests/src/bin/sequencer_node_end_to_end_integration_tests/integration_test_central_and_p2p_sync_flow.rs @@ -1,60 +1,6 @@ -use apollo_infra_utils::test_utils::TestIdentifier; -use apollo_integration_tests::integration_test_manager::IntegrationTestManager; -use apollo_integration_tests::integration_test_utils::integration_test_setup; -use apollo_integration_tests::utils::NodeDescriptor; -use apollo_node_config::component_execution_config::{ - ActiveComponentExecutionMode, - ReactiveComponentExecutionMode, -}; -use apollo_node_config::node_config::SequencerNodeConfig; -use apollo_state_sync_config::config::CentralSyncClientConfig; -use starknet_api::block::BlockNumber; -use tracing::info; +use apollo_integration_tests::flows; #[tokio::main] async fn main() { - integration_test_setup("sync").await; - const BLOCK_TO_WAIT_FOR: BlockNumber = BlockNumber(20); - // Node layout (index → type): 0 = consolidated, 1 = consolidated. - // Node 1 is configured as the central sync node. - const CENTRAL_SYNC_NODE: usize = 1; - - let node_descriptors = vec![NodeDescriptor::consolidated(), NodeDescriptor::consolidated()]; - - let mut integration_test_manager = IntegrationTestManager::new( - node_descriptors, - None, - TestIdentifier::SyncFlowIntegrationTest, - ) - .await; - - let update_config_disable_everything_but_sync = |config: &mut SequencerNodeConfig| { - config.components.batcher.execution_mode = ReactiveComponentExecutionMode::Disabled; - config.components.gateway.execution_mode = ReactiveComponentExecutionMode::Disabled; - config.components.mempool.execution_mode = ReactiveComponentExecutionMode::Disabled; - config.components.mempool_p2p.execution_mode = ReactiveComponentExecutionMode::Disabled; - config.components.l1_events_provider.execution_mode = - ReactiveComponentExecutionMode::Disabled; - config.components.consensus_manager.execution_mode = ActiveComponentExecutionMode::Disabled; - config.components.http_server.execution_mode = ActiveComponentExecutionMode::Disabled; - config.components.l1_events_scraper.execution_mode = ActiveComponentExecutionMode::Disabled; - }; - - let update_config_use_central_sync = |config: &mut SequencerNodeConfig| { - config.state_sync_config.as_mut().unwrap().static_config.central_sync_client_config = - Some(CentralSyncClientConfig::default()); - config.state_sync_config.as_mut().unwrap().static_config.p2p_sync_client_config = None; - }; - - let node_indices = integration_test_manager.get_node_indices(); - integration_test_manager - .modify_config_idle_nodes(node_indices.clone(), update_config_disable_everything_but_sync); - integration_test_manager - .modify_config_idle_nodes([CENTRAL_SYNC_NODE].into(), update_config_use_central_sync); - - integration_test_manager.run_nodes(node_indices.clone()).await; - - integration_test_manager.await_sync_block_on_all_running_nodes(BLOCK_TO_WAIT_FOR).await; - - info!("Sync flow integration test completed successfully!"); + flows::sync::run().await; } diff --git a/crates/apollo_integration_tests/src/bin/sequencer_node_end_to_end_integration_tests/integration_test_positive_flow.rs b/crates/apollo_integration_tests/src/bin/sequencer_node_end_to_end_integration_tests/integration_test_positive_flow.rs index 95a7f3d4479..19b3ff6f94b 100644 --- a/crates/apollo_integration_tests/src/bin/sequencer_node_end_to_end_integration_tests/integration_test_positive_flow.rs +++ b/crates/apollo_integration_tests/src/bin/sequencer_node_end_to_end_integration_tests/integration_test_positive_flow.rs @@ -1,57 +1,6 @@ -use apollo_infra_utils::test_utils::TestIdentifier; -use apollo_integration_tests::integration_test_manager::IntegrationTestManager; -use apollo_integration_tests::integration_test_utils::integration_test_setup; -use apollo_integration_tests::utils::NodeDescriptor; -use starknet_api::block::BlockNumber; -use tracing::info; +use apollo_integration_tests::flows; #[tokio::main] async fn main() { - integration_test_setup("positive").await; - const BLOCK_TO_WAIT_FOR: BlockNumber = BlockNumber(15); - const N_INVOKE_TXS: usize = 50; - const N_L1_HANDLER_TXS: usize = 2; - - let node_descriptors = vec![ - NodeDescriptor::consolidated(), - NodeDescriptor::consolidated(), - NodeDescriptor::consolidated(), - NodeDescriptor::distributed(), - NodeDescriptor::hybrid(), - ]; - - // Get the sequencer configurations. - let mut integration_test_manager = IntegrationTestManager::new( - node_descriptors, - None, - TestIdentifier::PositiveFlowIntegrationTest, - ) - .await; - - // TODO(Tsabary): consider decreasing - // "consensus_manager_config.consensus_manager_config.static_config.startup_delay" and - // "batcher_config.static_config.block_builder_config.proposer_idle_detection_delay_millis". - - let node_indices = integration_test_manager.get_node_indices(); - // Run the nodes. - integration_test_manager.run_nodes(node_indices.clone()).await; - - // Run the first block scenario to deploy the accounts. - integration_test_manager.send_deploy_and_invoke_txs_and_verify().await; - - integration_test_manager.send_declare_txs_and_verify().await; - - // Run the test. - integration_test_manager - .send_txs_and_verify(N_INVOKE_TXS, N_L1_HANDLER_TXS, BLOCK_TO_WAIT_FOR) - .await; - - integration_test_manager - .verify_block_hash_across_all_running_nodes(Some(BLOCK_TO_WAIT_FOR.unchecked_next())) - .await; - - info!("Shutting down nodes."); - integration_test_manager.shutdown_nodes(node_indices).await; - - info!("Positive flow integration test completed successfully!"); + flows::positive::run().await; } diff --git a/crates/apollo_integration_tests/src/bin/sequencer_node_end_to_end_integration_tests/integration_test_proof_flow.rs b/crates/apollo_integration_tests/src/bin/sequencer_node_end_to_end_integration_tests/integration_test_proof_flow.rs index f67848f048d..2e79eb3ebfb 100644 --- a/crates/apollo_integration_tests/src/bin/sequencer_node_end_to_end_integration_tests/integration_test_proof_flow.rs +++ b/crates/apollo_integration_tests/src/bin/sequencer_node_end_to_end_integration_tests/integration_test_proof_flow.rs @@ -1,54 +1,6 @@ -//! Integration test for the proof submission flow. -//! -//! The test relies on pre-generated fixture files in -//! `crates/apollo_integration_tests/resources/proof_flow/`. - -use apollo_infra_utils::test_utils::TestIdentifier; -use apollo_integration_tests::integration_test_manager::{ - IntegrationTestManager, - DEFAULT_SENDER_ACCOUNT, -}; -use apollo_integration_tests::integration_test_utils::integration_test_setup; -use apollo_integration_tests::utils::{NodeDescriptor, ProofFlowTxs}; -use blockifier::abi::constants::STORED_BLOCK_HASH_BUFFER; -use starknet_api::block::BlockNumber; -use tracing::info; +use apollo_integration_tests::flows; #[tokio::main] async fn main() { - integration_test_setup("proof_flow").await; - - // The fixture's proof references block 0, so the proof-bearing tx is only valid once the - // chain has progressed past `STORED_BLOCK_HASH_BUFFER`. We first advance the chain past that - // buffer with filler invokes, then submit the proof tx and wait one more block. - const BLOCK_PAST_HASH_BUFFER: BlockNumber = BlockNumber(STORED_BLOCK_HASH_BUFFER); - const BLOCK_TO_WAIT_FOR: BlockNumber = BlockNumber(STORED_BLOCK_HASH_BUFFER + 1); - let node_descriptors = vec![ - NodeDescriptor::consolidated(), - NodeDescriptor::distributed(), - NodeDescriptor::hybrid(), - ]; - - let mut integration_test_manager = IntegrationTestManager::new( - node_descriptors, - None, - TestIdentifier::ProofFlowIntegrationTest, - ) - .await; - - let node_indices = integration_test_manager.get_node_indices(); - integration_test_manager.run_nodes(node_indices.clone()).await; - - integration_test_manager.send_deploy_and_invoke_txs_and_verify().await; - - integration_test_manager.await_block_on_all_running_nodes(BLOCK_PAST_HASH_BUFFER).await; - - integration_test_manager - .test_and_verify(ProofFlowTxs::new(), DEFAULT_SENDER_ACCOUNT, BLOCK_TO_WAIT_FOR) - .await; - - info!("Shutting down nodes."); - integration_test_manager.shutdown_nodes(node_indices).await; - - info!("Proof flow integration test completed successfully!"); + flows::proof::run().await; } diff --git a/crates/apollo_integration_tests/src/bin/sequencer_node_end_to_end_integration_tests/integration_test_restart_flow.rs b/crates/apollo_integration_tests/src/bin/sequencer_node_end_to_end_integration_tests/integration_test_restart_flow.rs index a57aa066172..077c5bf5195 100644 --- a/crates/apollo_integration_tests/src/bin/sequencer_node_end_to_end_integration_tests/integration_test_restart_flow.rs +++ b/crates/apollo_integration_tests/src/bin/sequencer_node_end_to_end_integration_tests/integration_test_restart_flow.rs @@ -1,103 +1,6 @@ -use std::time::Duration; - -use apollo_deployments::service::NodeType; -use apollo_infra_utils::test_utils::TestIdentifier; -use apollo_integration_tests::integration_test_manager::{ - IntegrationTestManager, - DEFAULT_SENDER_ACCOUNT, -}; -use apollo_integration_tests::integration_test_utils::integration_test_setup; -use apollo_integration_tests::utils::NodeDescriptor; -use tracing::info; +use apollo_integration_tests::flows; #[tokio::main] async fn main() { - integration_test_setup("restart").await; - const TIMEOUT: Duration = Duration::from_secs(30); - const LONG_TIMEOUT: Duration = Duration::from_secs(90); - // Node layout (index → type): 0 = consolidated, 1 = hybrid, 2 = distributed. - // The test restarts a hybrid node and shuts down a non-consolidated (hybrid/distributed) node. - const RESTART_NODE: usize = 1; // hybrid - const SHUTDOWN_NODE: usize = 2; // distributed - - let node_descriptors = vec![ - NodeDescriptor::consolidated(), - NodeDescriptor::hybrid(), - NodeDescriptor::distributed(), - ]; - - // Get the sequencer configurations. - let mut integration_test_manager = IntegrationTestManager::new( - node_descriptors, - None, - TestIdentifier::RestartFlowIntegrationTest, - ) - .await; - - // Assert that RESTART_NODE is a hybrid node. - assert_eq!(integration_test_manager.get_node_type(RESTART_NODE), NodeType::Hybrid); - // Assert that SHUTDOWN_NODE is not a consolidated node. - assert_ne!(integration_test_manager.get_node_type(SHUTDOWN_NODE), NodeType::Consolidated); - - let mut node_indices = integration_test_manager.get_node_indices(); - - info!("Running all nodes."); - integration_test_manager.run_nodes(node_indices.clone()).await; - - integration_test_manager.send_deploy_and_invoke_txs_and_verify().await; - - integration_test_manager.send_declare_txs_and_verify().await; - - // Create a simulator for sustained transaction sending. - let simulator = integration_test_manager.create_simulator(); - let mut tx_generator = integration_test_manager.tx_generator().snapshot(); - - // Task that awaits transactions and restarts nodes in phases. - let await_and_restart_nodes_task = async { - info!("Awaiting transactions while all nodes are up"); - integration_test_manager.poll_all_running_nodes_received_more_txs(TIMEOUT).await; - - integration_test_manager.shutdown_nodes([RESTART_NODE].into()).await; - info!("Awaiting transactions while node {RESTART_NODE} is down"); - integration_test_manager.poll_all_running_nodes_received_more_txs(TIMEOUT).await; - - // We want the restarted node to rejoin the network while its building blocks to check the - // catch-up mechanism. - integration_test_manager.run_nodes([RESTART_NODE].into()).await; - info!( - "Awaiting node {RESTART_NODE} to join consensus after it was restarted and before \ - node {SHUTDOWN_NODE} is shut down" - ); - - integration_test_manager - .poll_node_reaches_consensus_decisions_after_restart(RESTART_NODE, LONG_TIMEOUT) - .await; - - integration_test_manager.poll_all_running_nodes_received_more_txs(TIMEOUT).await; - - // Shutdown a second node to test that the restarted node has joined consensus (the network - // can't reach consensus without the restarted node if the second node is down). - integration_test_manager.shutdown_nodes([SHUTDOWN_NODE].into()).await; - // Shutting down a node that's already down results in an error so we remove it from the set - // here. - node_indices.remove(&SHUTDOWN_NODE); - info!( - "Awaiting transactions while node {RESTART_NODE} is up and node {SHUTDOWN_NODE} is \ - down" - ); - integration_test_manager.poll_all_running_nodes_received_more_txs(LONG_TIMEOUT).await; - }; - - simulator - .run_test_with_nonstop_tx_sending( - &mut tx_generator, - DEFAULT_SENDER_ACCOUNT, - await_and_restart_nodes_task, - ) - .await; - - integration_test_manager.verify_block_hash_across_all_running_nodes(None).await; - - integration_test_manager.shutdown_nodes(node_indices).await; - info!("Restart flow integration test completed successfully!"); + flows::restart::run().await; } diff --git a/crates/apollo_integration_tests/src/bin/sequencer_node_end_to_end_integration_tests/integration_test_restart_service_multiple_nodes_flow.rs b/crates/apollo_integration_tests/src/bin/sequencer_node_end_to_end_integration_tests/integration_test_restart_service_multiple_nodes_flow.rs index 955b7e74480..503d8fd8d6c 100644 --- a/crates/apollo_integration_tests/src/bin/sequencer_node_end_to_end_integration_tests/integration_test_restart_service_multiple_nodes_flow.rs +++ b/crates/apollo_integration_tests/src/bin/sequencer_node_end_to_end_integration_tests/integration_test_restart_service_multiple_nodes_flow.rs @@ -1,86 +1,6 @@ -use std::collections::HashMap; - -use apollo_deployments::deployments::hybrid::HybridNodeServiceName; -use apollo_deployments::service::NodeService; -use apollo_infra_utils::test_utils::TestIdentifier; -use apollo_integration_tests::integration_test_manager::IntegrationTestManager; -use apollo_integration_tests::integration_test_utils::integration_test_setup; -use apollo_integration_tests::utils::NodeDescriptor; -use starknet_api::block::BlockNumber; -use strum::IntoEnumIterator; -use tracing::info; +use apollo_integration_tests::flows; #[tokio::main] async fn main() { - integration_test_setup("restart_service_multiple_nodes").await; - const INITIAL_BLOCK_TO_WAIT_FOR: usize = 20; - const BLOCK_TO_WAIT_FOR_INCREMENT: usize = 5; - const N_INVOKE_TXS: usize = 20; - const N_L1_HANDLER_TXS: usize = 1; - - let node_descriptors = vec![ - NodeDescriptor::hybrid(), - NodeDescriptor::hybrid(), - NodeDescriptor::hybrid(), - NodeDescriptor::hybrid(), - NodeDescriptor::hybrid(), - ]; - - // Get the sequencer configurations. - let mut integration_test_manager = IntegrationTestManager::new( - node_descriptors, - None, - TestIdentifier::RestartServiceMultipleNodesFlowIntegrationTest, - ) - .await; - - let node_indices = integration_test_manager.get_node_indices(); - // Run the nodes. - integration_test_manager.run_nodes(node_indices.clone()).await; - - // Run the first block scenario to deploy the accounts. - integration_test_manager.send_deploy_and_invoke_txs_and_verify().await; - - integration_test_manager.send_declare_txs_and_verify().await; - let mut block_to_wait_for = INITIAL_BLOCK_TO_WAIT_FOR; - for (i, hybrid_node_service) in HybridNodeServiceName::iter().enumerate() { - // TODO(noamsp): Remove this once the equivocaton feature is merged. - if hybrid_node_service == HybridNodeServiceName::Core { - continue; - } - - let node_services_to_shutdown = node_indices - .iter() - .map(|&node_index| (node_index, vec![hybrid_node_service.into()])) - .collect::>>(); - - info!("Shutting down service {hybrid_node_service:?} for all nodes."); - integration_test_manager.shutdown_node_services(node_services_to_shutdown.clone()); - info!("Running service {hybrid_node_service:?} for all nodes."); - integration_test_manager.run_node_services(node_services_to_shutdown).await; - - block_to_wait_for = INITIAL_BLOCK_TO_WAIT_FOR + i * BLOCK_TO_WAIT_FOR_INCREMENT; - info!( - "Sending txs and verifying after restarting service {hybrid_node_service:?} for all \ - nodes." - ); - integration_test_manager - .send_txs_and_verify( - N_INVOKE_TXS, - N_L1_HANDLER_TXS, - BlockNumber(block_to_wait_for.try_into().expect("Failed to convert to u64")), - ) - .await; - } - - integration_test_manager - .verify_block_hash_across_all_running_nodes(Some(BlockNumber( - (block_to_wait_for + 1).try_into().expect("Failed to convert to u64"), - ))) - .await; - - info!("Shutting down nodes."); - integration_test_manager.shutdown_nodes(node_indices).await; - - info!("Restart service multiple nodes flow integration test completed successfully!"); + flows::restart_multiple_nodes::run().await; } diff --git a/crates/apollo_integration_tests/src/bin/sequencer_node_end_to_end_integration_tests/integration_test_restart_service_single_node_flow.rs b/crates/apollo_integration_tests/src/bin/sequencer_node_end_to_end_integration_tests/integration_test_restart_service_single_node_flow.rs index e5f77fa7042..13b6eb21ac9 100644 --- a/crates/apollo_integration_tests/src/bin/sequencer_node_end_to_end_integration_tests/integration_test_restart_service_single_node_flow.rs +++ b/crates/apollo_integration_tests/src/bin/sequencer_node_end_to_end_integration_tests/integration_test_restart_service_single_node_flow.rs @@ -1,94 +1,6 @@ -use std::collections::HashMap; -use std::time::Duration; - -use apollo_deployments::deployments::hybrid::HybridNodeServiceName; -use apollo_deployments::service::NodeType; -use apollo_infra_utils::test_utils::TestIdentifier; -use apollo_integration_tests::integration_test_manager::{ - IntegrationTestManager, - DEFAULT_SENDER_ACCOUNT, -}; -use apollo_integration_tests::integration_test_utils::integration_test_setup; -use apollo_integration_tests::utils::NodeDescriptor; -use strum::IntoEnumIterator; -use tracing::info; +use apollo_integration_tests::flows; #[tokio::main] async fn main() { - integration_test_setup("restart_service_single_node").await; - const TIMEOUT: Duration = Duration::from_secs(30); - const LONG_TIMEOUT: Duration = Duration::from_secs(90); - // Node layout (index → type): 0 = hybrid, 1 = hybrid, 2 = hybrid. - // The test restarts the last hybrid node's service. - const RESTART_NODE: usize = 2; - - let node_descriptors = - vec![NodeDescriptor::hybrid(), NodeDescriptor::hybrid(), NodeDescriptor::hybrid()]; - - // Get the sequencer configurations. - let mut integration_test_manager = IntegrationTestManager::new( - node_descriptors, - None, - TestIdentifier::RestartServiceSingleNodeFlowIntegrationTest, - ) - .await; - - // Assert that RESTART_NODE is a hybrid node. - assert_eq!(integration_test_manager.get_node_type(RESTART_NODE), NodeType::Hybrid); - - let node_indices = integration_test_manager.get_node_indices(); - integration_test_manager.run_nodes(node_indices.clone()).await; - - integration_test_manager.send_deploy_and_invoke_txs_and_verify().await; - - integration_test_manager.send_declare_txs_and_verify().await; - - // Create a simulator for sustained transaction sending. - let simulator = integration_test_manager.create_simulator(); - let mut tx_generator = integration_test_manager.tx_generator().snapshot(); - - // The indices of the nodes that are healthy throughout the test. - let mut healthy_node_indices = node_indices.clone(); - healthy_node_indices.remove(&RESTART_NODE); - // Task that awaits transactions and restarts nodes in phases. - let await_and_restart_nodes_task = async { - for hybrid_node_service in HybridNodeServiceName::iter() { - // TODO(noamsp): Remove this once the equivocaton feature is merged. - if hybrid_node_service == HybridNodeServiceName::Core { - continue; - } - - info!("Shutting down service {hybrid_node_service:?} for node {RESTART_NODE}."); - let restart_node_service = - HashMap::from([(RESTART_NODE, vec![hybrid_node_service.into()])]); - integration_test_manager.shutdown_node_services(restart_node_service.clone()); - - // Verify that the other nodes are still running properly. - integration_test_manager - .poll_running_nodes_received_more_txs(TIMEOUT, &healthy_node_indices) - .await; - - info!("Running service {hybrid_node_service:?} for node {RESTART_NODE}."); - integration_test_manager.run_node_services(restart_node_service.clone()).await; - - integration_test_manager - .poll_node_reaches_consensus_decisions_after_restart(RESTART_NODE, LONG_TIMEOUT) - .await; - - integration_test_manager.poll_all_running_nodes_received_more_txs(LONG_TIMEOUT).await; - } - }; - - simulator - .run_test_with_nonstop_tx_sending( - &mut tx_generator, - DEFAULT_SENDER_ACCOUNT, - await_and_restart_nodes_task, - ) - .await; - - integration_test_manager.verify_block_hash_across_all_running_nodes(None).await; - - integration_test_manager.shutdown_nodes(node_indices).await; - info!("Restart service single node flow integration test completed successfully!"); + flows::restart_single_node::run().await; } diff --git a/crates/apollo_integration_tests/src/bin/sequencer_node_end_to_end_integration_tests/integration_test_revert_flow.rs b/crates/apollo_integration_tests/src/bin/sequencer_node_end_to_end_integration_tests/integration_test_revert_flow.rs index edf247a3b63..16fa903dded 100644 --- a/crates/apollo_integration_tests/src/bin/sequencer_node_end_to_end_integration_tests/integration_test_revert_flow.rs +++ b/crates/apollo_integration_tests/src/bin/sequencer_node_end_to_end_integration_tests/integration_test_revert_flow.rs @@ -1,209 +1,6 @@ -use std::collections::HashSet; -use std::time::Duration; - -use apollo_infra_utils::test_utils::TestIdentifier; -use apollo_integration_tests::integration_test_manager::IntegrationTestManager; -use apollo_integration_tests::integration_test_utils::integration_test_setup; -use apollo_integration_tests::utils::NodeDescriptor; -use apollo_node_config::definitions::ConfigPointersMap; -use apollo_node_config::node_config::SequencerNodeConfig; -use serde_json::Value; -use starknet_api::block::BlockNumber; -use tracing::info; +use apollo_integration_tests::flows; #[tokio::main] async fn main() { - integration_test_setup("revert").await; - const BLOCK_TO_REVERT_FROM: BlockNumber = BlockNumber(30); - const REVERT_UP_TO_AND_INCLUDING: BlockNumber = BlockNumber(1); - const BLOCK_TO_WAIT_FOR_AFTER_REVERT: BlockNumber = BlockNumber(40); - // can't use static assertion as comparison is non const. - assert!(REVERT_UP_TO_AND_INCLUDING < BLOCK_TO_REVERT_FROM); - assert!(BLOCK_TO_REVERT_FROM < BLOCK_TO_WAIT_FOR_AFTER_REVERT); - - const N_INVOKE_TXS: usize = 50; - const N_L1_HANDLER_TXS: usize = 5; - - const AWAIT_REVERT_INTERVAL_MS: u64 = 500; - const MAX_ATTEMPTS: usize = 50; - const AWAIT_REVERT_TIMEOUT_DURATION: Duration = Duration::from_secs(15); - - let node_descriptors = vec![ - NodeDescriptor::consolidated(), - NodeDescriptor::consolidated(), - NodeDescriptor::consolidated(), - NodeDescriptor::consolidated(), - NodeDescriptor::consolidated(), - ]; - - // Get the sequencer configurations. - let mut integration_test_manager = IntegrationTestManager::new( - node_descriptors, - None, - TestIdentifier::RevertFlowIntegrationTest, - ) - .await; - - let node_indices = integration_test_manager.get_node_indices(); - - integration_test_manager.run_nodes(node_indices.clone()).await; - - // Save a snapshot of the tx_generator so we can restore the state after reverting. - let tx_generator_snapshot = integration_test_manager.tx_generator().snapshot(); - - info!("Sending deploy and invoke together transactions and verifying state."); - integration_test_manager.send_deploy_and_invoke_txs_and_verify().await; - - info!("Sending declare transactions and verifying state."); - integration_test_manager.send_declare_txs_and_verify().await; - - info!("Sending transactions and verifying state."); - integration_test_manager - .send_txs_and_verify(N_INVOKE_TXS, N_L1_HANDLER_TXS, BLOCK_TO_REVERT_FROM) - .await; - - // This method is called twice in the test, once before and once after the revert. - // The first call is mainly to verify block hashes were written to storage before the revert. - integration_test_manager - .verify_block_hash_across_all_running_nodes(Some(BLOCK_TO_REVERT_FROM.unchecked_next())) - .await; - - info!("Shutting down nodes."); - integration_test_manager.shutdown_nodes(node_indices.clone()).await; - - let expected_block_number_after_revert = REVERT_UP_TO_AND_INCLUDING.prev().unwrap_or_default(); - info!( - "Changing revert config for all nodes to revert from block {BLOCK_TO_REVERT_FROM} back to \ - block {expected_block_number_after_revert}." - ); - modify_revert_config_idle_nodes( - &mut integration_test_manager, - node_indices.clone(), - Some(REVERT_UP_TO_AND_INCLUDING), - ); - - integration_test_manager.run_nodes(node_indices.clone()).await; - - info!( - "Awaiting for all running nodes to revert back to block \ - {expected_block_number_after_revert}.", - ); - integration_test_manager - .await_revert_all_running_nodes( - expected_block_number_after_revert, - AWAIT_REVERT_TIMEOUT_DURATION, - AWAIT_REVERT_INTERVAL_MS, - MAX_ATTEMPTS, - ) - .await; - - info!("All nodes reverted to block {expected_block_number_after_revert}. Shutting down nodes."); - integration_test_manager.shutdown_nodes(node_indices.clone()).await; - - // Restore the tx generator state. - *integration_test_manager.tx_generator_mut() = tx_generator_snapshot; - - info!( - "Modifying revert config for all nodes and resume sequencing from block \ - {expected_block_number_after_revert}." - ); - modify_revert_config_idle_nodes(&mut integration_test_manager, node_indices.clone(), None); - modify_height_configs_idle_nodes(&mut integration_test_manager, node_indices.clone()); - - integration_test_manager.run_nodes(node_indices.clone()).await; - - info!("Sending deploy and invoke together transactions and verifying state."); - integration_test_manager.send_deploy_and_invoke_txs_and_verify().await; - - info!("Sending declare transactions and verifying state."); - integration_test_manager.send_declare_txs_and_verify().await; - - info!("Sending transactions and verifying state."); - integration_test_manager - .send_txs_and_verify(N_INVOKE_TXS, N_L1_HANDLER_TXS, BLOCK_TO_WAIT_FOR_AFTER_REVERT) - .await; - - integration_test_manager - .verify_block_hash_across_all_running_nodes(Some( - BLOCK_TO_WAIT_FOR_AFTER_REVERT.unchecked_next(), - )) - .await; - - integration_test_manager.shutdown_nodes(node_indices).await; - - info!("Revert flow integration test completed successfully!"); -} - -// Modifies the revert config state in the given config. If `revert_up_to_and_including` is -// `None`, the revert config is disabled. Otherwise, the revert config is enabled and set -// to revert up to and including the given block number. -fn modify_revert_config_idle_nodes( - integration_test_manager: &mut IntegrationTestManager, - node_indices: HashSet, - revert_up_to_and_including: Option, -) { - integration_test_manager.modify_config_pointers_idle_nodes( - node_indices.clone(), - |config_pointers| { - modify_revert_config_pointers(config_pointers, revert_up_to_and_including) - }, - ); - integration_test_manager.modify_config_idle_nodes(node_indices, |config_pointers| { - modify_revert_config(config_pointers, revert_up_to_and_including) - }); -} - -fn modify_revert_config_pointers( - config_pointers: &mut ConfigPointersMap, - revert_up_to_and_including: Option, -) { - let should_revert = revert_up_to_and_including.is_some(); - config_pointers.change_target_value("revert_config.should_revert", Value::from(should_revert)); - - // If should revert is false, the revert_up_to_and_including value is irrelevant. - if should_revert { - let revert_up_to_and_including = revert_up_to_and_including.unwrap(); - config_pointers.change_target_value( - "revert_config.revert_up_to_and_including", - Value::from(revert_up_to_and_including.0), - ); - } -} - -fn modify_revert_config( - config: &mut SequencerNodeConfig, - revert_up_to_and_including: Option, -) { - let should_revert = revert_up_to_and_including.is_some(); - config.state_sync_config.as_mut().unwrap().static_config.revert_config.should_revert = - should_revert; - config.consensus_manager_config.as_mut().unwrap().revert_config.should_revert = should_revert; - - // If should revert is false, the revert_up_to_and_including value is irrelevant. - if should_revert { - let revert_up_to_and_including = revert_up_to_and_including.unwrap(); - config - .state_sync_config - .as_mut() - .unwrap() - .static_config - .revert_config - .revert_up_to_and_including = revert_up_to_and_including; - config - .consensus_manager_config - .as_mut() - .unwrap() - .revert_config - .revert_up_to_and_including = revert_up_to_and_including; - } -} - -fn modify_height_configs_idle_nodes( - integration_test_manager: &mut IntegrationTestManager, - node_indices: HashSet, -) { - integration_test_manager.modify_config_idle_nodes(node_indices, |_config| { - // TODO(noamsp): Change these values point to a single config value and refactor this - // function accordingly. - }); + flows::revert::run().await; } diff --git a/crates/apollo_integration_tests/src/flows/mod.rs b/crates/apollo_integration_tests/src/flows/mod.rs new file mode 100644 index 00000000000..1deff4676e4 --- /dev/null +++ b/crates/apollo_integration_tests/src/flows/mod.rs @@ -0,0 +1,10 @@ +//! End-to-end integration-test flows, one module per flow. Each flow orchestrates sequencer +//! nodes (spawned as child processes) end to end and is designed to run as its own process. + +pub mod positive; +pub mod proof; +pub mod restart; +pub mod restart_multiple_nodes; +pub mod restart_single_node; +pub mod revert; +pub mod sync; diff --git a/crates/apollo_integration_tests/src/flows/positive.rs b/crates/apollo_integration_tests/src/flows/positive.rs new file mode 100644 index 00000000000..4b5832751ba --- /dev/null +++ b/crates/apollo_integration_tests/src/flows/positive.rs @@ -0,0 +1,57 @@ +use apollo_infra_utils::test_utils::TestIdentifier; +use starknet_api::block::BlockNumber; +use tracing::info; + +use crate::integration_test_manager::IntegrationTestManager; +use crate::integration_test_utils::integration_test_setup; +use crate::utils::NodeDescriptor; + +pub async fn run() { + integration_test_setup("positive").await; + const BLOCK_TO_WAIT_FOR: BlockNumber = BlockNumber(15); + const N_INVOKE_TXS: usize = 50; + const N_L1_HANDLER_TXS: usize = 2; + + let node_descriptors = vec![ + NodeDescriptor::consolidated(), + NodeDescriptor::consolidated(), + NodeDescriptor::consolidated(), + NodeDescriptor::distributed(), + NodeDescriptor::hybrid(), + ]; + + // Get the sequencer configurations. + let mut integration_test_manager = IntegrationTestManager::new( + node_descriptors, + None, + TestIdentifier::PositiveFlowIntegrationTest, + ) + .await; + + // TODO(Tsabary): consider decreasing + // "consensus_manager_config.consensus_manager_config.static_config.startup_delay" and + // "batcher_config.static_config.block_builder_config.proposer_idle_detection_delay_millis". + + let node_indices = integration_test_manager.get_node_indices(); + // Run the nodes. + integration_test_manager.run_nodes(node_indices.clone()).await; + + // Run the first block scenario to deploy the accounts. + integration_test_manager.send_deploy_and_invoke_txs_and_verify().await; + + integration_test_manager.send_declare_txs_and_verify().await; + + // Run the test. + integration_test_manager + .send_txs_and_verify(N_INVOKE_TXS, N_L1_HANDLER_TXS, BLOCK_TO_WAIT_FOR) + .await; + + integration_test_manager + .verify_block_hash_across_all_running_nodes(Some(BLOCK_TO_WAIT_FOR.unchecked_next())) + .await; + + info!("Shutting down nodes."); + integration_test_manager.shutdown_nodes(node_indices).await; + + info!("Positive flow integration test completed successfully!"); +} diff --git a/crates/apollo_integration_tests/src/flows/proof.rs b/crates/apollo_integration_tests/src/flows/proof.rs new file mode 100644 index 00000000000..7bdf943e7a9 --- /dev/null +++ b/crates/apollo_integration_tests/src/flows/proof.rs @@ -0,0 +1,51 @@ +//! Integration test for the proof submission flow. +//! +//! The test relies on pre-generated fixture files in +//! `crates/apollo_integration_tests/resources/proof_flow/`. + +use apollo_infra_utils::test_utils::TestIdentifier; +use blockifier::abi::constants::STORED_BLOCK_HASH_BUFFER; +use starknet_api::block::BlockNumber; +use tracing::info; + +use crate::integration_test_manager::{IntegrationTestManager, DEFAULT_SENDER_ACCOUNT}; +use crate::integration_test_utils::integration_test_setup; +use crate::utils::{NodeDescriptor, ProofFlowTxs}; + +pub async fn run() { + integration_test_setup("proof_flow").await; + + // The fixture's proof references block 0, so the proof-bearing tx is only valid once the + // chain has progressed past `STORED_BLOCK_HASH_BUFFER`. We first advance the chain past that + // buffer with filler invokes, then submit the proof tx and wait one more block. + const BLOCK_PAST_HASH_BUFFER: BlockNumber = BlockNumber(STORED_BLOCK_HASH_BUFFER); + const BLOCK_TO_WAIT_FOR: BlockNumber = BlockNumber(STORED_BLOCK_HASH_BUFFER + 1); + let node_descriptors = vec![ + NodeDescriptor::consolidated(), + NodeDescriptor::distributed(), + NodeDescriptor::hybrid(), + ]; + + let mut integration_test_manager = IntegrationTestManager::new( + node_descriptors, + None, + TestIdentifier::ProofFlowIntegrationTest, + ) + .await; + + let node_indices = integration_test_manager.get_node_indices(); + integration_test_manager.run_nodes(node_indices.clone()).await; + + integration_test_manager.send_deploy_and_invoke_txs_and_verify().await; + + integration_test_manager.await_block_on_all_running_nodes(BLOCK_PAST_HASH_BUFFER).await; + + integration_test_manager + .test_and_verify(ProofFlowTxs::new(), DEFAULT_SENDER_ACCOUNT, BLOCK_TO_WAIT_FOR) + .await; + + info!("Shutting down nodes."); + integration_test_manager.shutdown_nodes(node_indices).await; + + info!("Proof flow integration test completed successfully!"); +} diff --git a/crates/apollo_integration_tests/src/flows/restart.rs b/crates/apollo_integration_tests/src/flows/restart.rs new file mode 100644 index 00000000000..6e4d59e4b25 --- /dev/null +++ b/crates/apollo_integration_tests/src/flows/restart.rs @@ -0,0 +1,100 @@ +use std::time::Duration; + +use apollo_deployments::service::NodeType; +use apollo_infra_utils::test_utils::TestIdentifier; +use tracing::info; + +use crate::integration_test_manager::{IntegrationTestManager, DEFAULT_SENDER_ACCOUNT}; +use crate::integration_test_utils::integration_test_setup; +use crate::utils::NodeDescriptor; + +pub async fn run() { + integration_test_setup("restart").await; + const TIMEOUT: Duration = Duration::from_secs(30); + const LONG_TIMEOUT: Duration = Duration::from_secs(90); + // Node layout (index → type): 0 = consolidated, 1 = hybrid, 2 = distributed. + // The test restarts a hybrid node and shuts down a non-consolidated (hybrid/distributed) node. + const RESTART_NODE: usize = 1; // hybrid + const SHUTDOWN_NODE: usize = 2; // distributed + + let node_descriptors = vec![ + NodeDescriptor::consolidated(), + NodeDescriptor::hybrid(), + NodeDescriptor::distributed(), + ]; + + // Get the sequencer configurations. + let mut integration_test_manager = IntegrationTestManager::new( + node_descriptors, + None, + TestIdentifier::RestartFlowIntegrationTest, + ) + .await; + + // Assert that RESTART_NODE is a hybrid node. + assert_eq!(integration_test_manager.get_node_type(RESTART_NODE), NodeType::Hybrid); + // Assert that SHUTDOWN_NODE is not a consolidated node. + assert_ne!(integration_test_manager.get_node_type(SHUTDOWN_NODE), NodeType::Consolidated); + + let mut node_indices = integration_test_manager.get_node_indices(); + + info!("Running all nodes."); + integration_test_manager.run_nodes(node_indices.clone()).await; + + integration_test_manager.send_deploy_and_invoke_txs_and_verify().await; + + integration_test_manager.send_declare_txs_and_verify().await; + + // Create a simulator for sustained transaction sending. + let simulator = integration_test_manager.create_simulator(); + let mut tx_generator = integration_test_manager.tx_generator().snapshot(); + + // Task that awaits transactions and restarts nodes in phases. + let await_and_restart_nodes_task = async { + info!("Awaiting transactions while all nodes are up"); + integration_test_manager.poll_all_running_nodes_received_more_txs(TIMEOUT).await; + + integration_test_manager.shutdown_nodes([RESTART_NODE].into()).await; + info!("Awaiting transactions while node {RESTART_NODE} is down"); + integration_test_manager.poll_all_running_nodes_received_more_txs(TIMEOUT).await; + + // We want the restarted node to rejoin the network while its building blocks to check the + // catch-up mechanism. + integration_test_manager.run_nodes([RESTART_NODE].into()).await; + info!( + "Awaiting node {RESTART_NODE} to join consensus after it was restarted and before \ + node {SHUTDOWN_NODE} is shut down" + ); + + integration_test_manager + .poll_node_reaches_consensus_decisions_after_restart(RESTART_NODE, LONG_TIMEOUT) + .await; + + integration_test_manager.poll_all_running_nodes_received_more_txs(TIMEOUT).await; + + // Shutdown a second node to test that the restarted node has joined consensus (the network + // can't reach consensus without the restarted node if the second node is down). + integration_test_manager.shutdown_nodes([SHUTDOWN_NODE].into()).await; + // Shutting down a node that's already down results in an error so we remove it from the set + // here. + node_indices.remove(&SHUTDOWN_NODE); + info!( + "Awaiting transactions while node {RESTART_NODE} is up and node {SHUTDOWN_NODE} is \ + down" + ); + integration_test_manager.poll_all_running_nodes_received_more_txs(LONG_TIMEOUT).await; + }; + + simulator + .run_test_with_nonstop_tx_sending( + &mut tx_generator, + DEFAULT_SENDER_ACCOUNT, + await_and_restart_nodes_task, + ) + .await; + + integration_test_manager.verify_block_hash_across_all_running_nodes(None).await; + + integration_test_manager.shutdown_nodes(node_indices).await; + info!("Restart flow integration test completed successfully!"); +} diff --git a/crates/apollo_integration_tests/src/flows/restart_multiple_nodes.rs b/crates/apollo_integration_tests/src/flows/restart_multiple_nodes.rs new file mode 100644 index 00000000000..10dcb6ae1fe --- /dev/null +++ b/crates/apollo_integration_tests/src/flows/restart_multiple_nodes.rs @@ -0,0 +1,86 @@ +use std::collections::HashMap; + +use apollo_deployments::deployments::hybrid::HybridNodeServiceName; +use apollo_deployments::service::NodeService; +use apollo_infra_utils::test_utils::TestIdentifier; +use starknet_api::block::BlockNumber; +use strum::IntoEnumIterator; +use tracing::info; + +use crate::integration_test_manager::IntegrationTestManager; +use crate::integration_test_utils::integration_test_setup; +use crate::utils::NodeDescriptor; + +pub async fn run() { + integration_test_setup("restart_service_multiple_nodes").await; + const INITIAL_BLOCK_TO_WAIT_FOR: usize = 20; + const BLOCK_TO_WAIT_FOR_INCREMENT: usize = 5; + const N_INVOKE_TXS: usize = 20; + const N_L1_HANDLER_TXS: usize = 1; + + let node_descriptors = vec![ + NodeDescriptor::hybrid(), + NodeDescriptor::hybrid(), + NodeDescriptor::hybrid(), + NodeDescriptor::hybrid(), + NodeDescriptor::hybrid(), + ]; + + // Get the sequencer configurations. + let mut integration_test_manager = IntegrationTestManager::new( + node_descriptors, + None, + TestIdentifier::RestartServiceMultipleNodesFlowIntegrationTest, + ) + .await; + + let node_indices = integration_test_manager.get_node_indices(); + // Run the nodes. + integration_test_manager.run_nodes(node_indices.clone()).await; + + // Run the first block scenario to deploy the accounts. + integration_test_manager.send_deploy_and_invoke_txs_and_verify().await; + + integration_test_manager.send_declare_txs_and_verify().await; + let mut block_to_wait_for = INITIAL_BLOCK_TO_WAIT_FOR; + for (i, hybrid_node_service) in HybridNodeServiceName::iter().enumerate() { + // TODO(noamsp): Remove this once the equivocaton feature is merged. + if hybrid_node_service == HybridNodeServiceName::Core { + continue; + } + + let node_services_to_shutdown = node_indices + .iter() + .map(|&node_index| (node_index, vec![hybrid_node_service.into()])) + .collect::>>(); + + info!("Shutting down service {hybrid_node_service:?} for all nodes."); + integration_test_manager.shutdown_node_services(node_services_to_shutdown.clone()); + info!("Running service {hybrid_node_service:?} for all nodes."); + integration_test_manager.run_node_services(node_services_to_shutdown).await; + + block_to_wait_for = INITIAL_BLOCK_TO_WAIT_FOR + i * BLOCK_TO_WAIT_FOR_INCREMENT; + info!( + "Sending txs and verifying after restarting service {hybrid_node_service:?} for all \ + nodes." + ); + integration_test_manager + .send_txs_and_verify( + N_INVOKE_TXS, + N_L1_HANDLER_TXS, + BlockNumber(block_to_wait_for.try_into().expect("Failed to convert to u64")), + ) + .await; + } + + integration_test_manager + .verify_block_hash_across_all_running_nodes(Some(BlockNumber( + (block_to_wait_for + 1).try_into().expect("Failed to convert to u64"), + ))) + .await; + + info!("Shutting down nodes."); + integration_test_manager.shutdown_nodes(node_indices).await; + + info!("Restart service multiple nodes flow integration test completed successfully!"); +} diff --git a/crates/apollo_integration_tests/src/flows/restart_single_node.rs b/crates/apollo_integration_tests/src/flows/restart_single_node.rs new file mode 100644 index 00000000000..b487703732d --- /dev/null +++ b/crates/apollo_integration_tests/src/flows/restart_single_node.rs @@ -0,0 +1,91 @@ +use std::collections::HashMap; +use std::time::Duration; + +use apollo_deployments::deployments::hybrid::HybridNodeServiceName; +use apollo_deployments::service::NodeType; +use apollo_infra_utils::test_utils::TestIdentifier; +use strum::IntoEnumIterator; +use tracing::info; + +use crate::integration_test_manager::{IntegrationTestManager, DEFAULT_SENDER_ACCOUNT}; +use crate::integration_test_utils::integration_test_setup; +use crate::utils::NodeDescriptor; + +pub async fn run() { + integration_test_setup("restart_service_single_node").await; + const TIMEOUT: Duration = Duration::from_secs(30); + const LONG_TIMEOUT: Duration = Duration::from_secs(90); + // Node layout (index → type): 0 = hybrid, 1 = hybrid, 2 = hybrid. + // The test restarts the last hybrid node's service. + const RESTART_NODE: usize = 2; + + let node_descriptors = + vec![NodeDescriptor::hybrid(), NodeDescriptor::hybrid(), NodeDescriptor::hybrid()]; + + // Get the sequencer configurations. + let mut integration_test_manager = IntegrationTestManager::new( + node_descriptors, + None, + TestIdentifier::RestartServiceSingleNodeFlowIntegrationTest, + ) + .await; + + // Assert that RESTART_NODE is a hybrid node. + assert_eq!(integration_test_manager.get_node_type(RESTART_NODE), NodeType::Hybrid); + + let node_indices = integration_test_manager.get_node_indices(); + integration_test_manager.run_nodes(node_indices.clone()).await; + + integration_test_manager.send_deploy_and_invoke_txs_and_verify().await; + + integration_test_manager.send_declare_txs_and_verify().await; + + // Create a simulator for sustained transaction sending. + let simulator = integration_test_manager.create_simulator(); + let mut tx_generator = integration_test_manager.tx_generator().snapshot(); + + // The indices of the nodes that are healthy throughout the test. + let mut healthy_node_indices = node_indices.clone(); + healthy_node_indices.remove(&RESTART_NODE); + // Task that awaits transactions and restarts nodes in phases. + let await_and_restart_nodes_task = async { + for hybrid_node_service in HybridNodeServiceName::iter() { + // TODO(noamsp): Remove this once the equivocaton feature is merged. + if hybrid_node_service == HybridNodeServiceName::Core { + continue; + } + + info!("Shutting down service {hybrid_node_service:?} for node {RESTART_NODE}."); + let restart_node_service = + HashMap::from([(RESTART_NODE, vec![hybrid_node_service.into()])]); + integration_test_manager.shutdown_node_services(restart_node_service.clone()); + + // Verify that the other nodes are still running properly. + integration_test_manager + .poll_running_nodes_received_more_txs(TIMEOUT, &healthy_node_indices) + .await; + + info!("Running service {hybrid_node_service:?} for node {RESTART_NODE}."); + integration_test_manager.run_node_services(restart_node_service.clone()).await; + + integration_test_manager + .poll_node_reaches_consensus_decisions_after_restart(RESTART_NODE, LONG_TIMEOUT) + .await; + + integration_test_manager.poll_all_running_nodes_received_more_txs(LONG_TIMEOUT).await; + } + }; + + simulator + .run_test_with_nonstop_tx_sending( + &mut tx_generator, + DEFAULT_SENDER_ACCOUNT, + await_and_restart_nodes_task, + ) + .await; + + integration_test_manager.verify_block_hash_across_all_running_nodes(None).await; + + integration_test_manager.shutdown_nodes(node_indices).await; + info!("Restart service single node flow integration test completed successfully!"); +} diff --git a/crates/apollo_integration_tests/src/flows/revert.rs b/crates/apollo_integration_tests/src/flows/revert.rs new file mode 100644 index 00000000000..fa6aeb5e56d --- /dev/null +++ b/crates/apollo_integration_tests/src/flows/revert.rs @@ -0,0 +1,209 @@ +use std::collections::HashSet; +use std::time::Duration; + +use apollo_infra_utils::test_utils::TestIdentifier; +use apollo_node_config::definitions::ConfigPointersMap; +use apollo_node_config::node_config::SequencerNodeConfig; +use serde_json::Value; +use starknet_api::block::BlockNumber; +use tracing::info; + +use crate::integration_test_manager::IntegrationTestManager; +use crate::integration_test_utils::integration_test_setup; +use crate::utils::NodeDescriptor; + +pub async fn run() { + integration_test_setup("revert").await; + const BLOCK_TO_REVERT_FROM: BlockNumber = BlockNumber(30); + const REVERT_UP_TO_AND_INCLUDING: BlockNumber = BlockNumber(1); + const BLOCK_TO_WAIT_FOR_AFTER_REVERT: BlockNumber = BlockNumber(40); + // can't use static assertion as comparison is non const. + assert!(REVERT_UP_TO_AND_INCLUDING < BLOCK_TO_REVERT_FROM); + assert!(BLOCK_TO_REVERT_FROM < BLOCK_TO_WAIT_FOR_AFTER_REVERT); + + const N_INVOKE_TXS: usize = 50; + const N_L1_HANDLER_TXS: usize = 5; + + const AWAIT_REVERT_INTERVAL_MS: u64 = 500; + const MAX_ATTEMPTS: usize = 50; + const AWAIT_REVERT_TIMEOUT_DURATION: Duration = Duration::from_secs(15); + + let node_descriptors = vec![ + NodeDescriptor::consolidated(), + NodeDescriptor::consolidated(), + NodeDescriptor::consolidated(), + NodeDescriptor::consolidated(), + NodeDescriptor::consolidated(), + ]; + + // Get the sequencer configurations. + let mut integration_test_manager = IntegrationTestManager::new( + node_descriptors, + None, + TestIdentifier::RevertFlowIntegrationTest, + ) + .await; + + let node_indices = integration_test_manager.get_node_indices(); + + integration_test_manager.run_nodes(node_indices.clone()).await; + + // Save a snapshot of the tx_generator so we can restore the state after reverting. + let tx_generator_snapshot = integration_test_manager.tx_generator().snapshot(); + + info!("Sending deploy and invoke together transactions and verifying state."); + integration_test_manager.send_deploy_and_invoke_txs_and_verify().await; + + info!("Sending declare transactions and verifying state."); + integration_test_manager.send_declare_txs_and_verify().await; + + info!("Sending transactions and verifying state."); + integration_test_manager + .send_txs_and_verify(N_INVOKE_TXS, N_L1_HANDLER_TXS, BLOCK_TO_REVERT_FROM) + .await; + + // This method is called twice in the test, once before and once after the revert. + // The first call is mainly to verify block hashes were written to storage before the revert. + integration_test_manager + .verify_block_hash_across_all_running_nodes(Some(BLOCK_TO_REVERT_FROM.unchecked_next())) + .await; + + info!("Shutting down nodes."); + integration_test_manager.shutdown_nodes(node_indices.clone()).await; + + let expected_block_number_after_revert = REVERT_UP_TO_AND_INCLUDING.prev().unwrap_or_default(); + info!( + "Changing revert config for all nodes to revert from block {BLOCK_TO_REVERT_FROM} back to \ + block {expected_block_number_after_revert}." + ); + modify_revert_config_idle_nodes( + &mut integration_test_manager, + node_indices.clone(), + Some(REVERT_UP_TO_AND_INCLUDING), + ); + + integration_test_manager.run_nodes(node_indices.clone()).await; + + info!( + "Awaiting for all running nodes to revert back to block \ + {expected_block_number_after_revert}.", + ); + integration_test_manager + .await_revert_all_running_nodes( + expected_block_number_after_revert, + AWAIT_REVERT_TIMEOUT_DURATION, + AWAIT_REVERT_INTERVAL_MS, + MAX_ATTEMPTS, + ) + .await; + + info!("All nodes reverted to block {expected_block_number_after_revert}. Shutting down nodes."); + integration_test_manager.shutdown_nodes(node_indices.clone()).await; + + // Restore the tx generator state. + *integration_test_manager.tx_generator_mut() = tx_generator_snapshot; + + info!( + "Modifying revert config for all nodes and resume sequencing from block \ + {expected_block_number_after_revert}." + ); + modify_revert_config_idle_nodes(&mut integration_test_manager, node_indices.clone(), None); + modify_height_configs_idle_nodes(&mut integration_test_manager, node_indices.clone()); + + integration_test_manager.run_nodes(node_indices.clone()).await; + + info!("Sending deploy and invoke together transactions and verifying state."); + integration_test_manager.send_deploy_and_invoke_txs_and_verify().await; + + info!("Sending declare transactions and verifying state."); + integration_test_manager.send_declare_txs_and_verify().await; + + info!("Sending transactions and verifying state."); + integration_test_manager + .send_txs_and_verify(N_INVOKE_TXS, N_L1_HANDLER_TXS, BLOCK_TO_WAIT_FOR_AFTER_REVERT) + .await; + + integration_test_manager + .verify_block_hash_across_all_running_nodes(Some( + BLOCK_TO_WAIT_FOR_AFTER_REVERT.unchecked_next(), + )) + .await; + + integration_test_manager.shutdown_nodes(node_indices).await; + + info!("Revert flow integration test completed successfully!"); +} + +// Modifies the revert config state in the given config. If `revert_up_to_and_including` is +// `None`, the revert config is disabled. Otherwise, the revert config is enabled and set +// to revert up to and including the given block number. +fn modify_revert_config_idle_nodes( + integration_test_manager: &mut IntegrationTestManager, + node_indices: HashSet, + revert_up_to_and_including: Option, +) { + integration_test_manager.modify_config_pointers_idle_nodes( + node_indices.clone(), + |config_pointers| { + modify_revert_config_pointers(config_pointers, revert_up_to_and_including) + }, + ); + integration_test_manager.modify_config_idle_nodes(node_indices, |config_pointers| { + modify_revert_config(config_pointers, revert_up_to_and_including) + }); +} + +fn modify_revert_config_pointers( + config_pointers: &mut ConfigPointersMap, + revert_up_to_and_including: Option, +) { + let should_revert = revert_up_to_and_including.is_some(); + config_pointers.change_target_value("revert_config.should_revert", Value::from(should_revert)); + + // If should revert is false, the revert_up_to_and_including value is irrelevant. + if should_revert { + let revert_up_to_and_including = revert_up_to_and_including.unwrap(); + config_pointers.change_target_value( + "revert_config.revert_up_to_and_including", + Value::from(revert_up_to_and_including.0), + ); + } +} + +fn modify_revert_config( + config: &mut SequencerNodeConfig, + revert_up_to_and_including: Option, +) { + let should_revert = revert_up_to_and_including.is_some(); + config.state_sync_config.as_mut().unwrap().static_config.revert_config.should_revert = + should_revert; + config.consensus_manager_config.as_mut().unwrap().revert_config.should_revert = should_revert; + + // If should revert is false, the revert_up_to_and_including value is irrelevant. + if should_revert { + let revert_up_to_and_including = revert_up_to_and_including.unwrap(); + config + .state_sync_config + .as_mut() + .unwrap() + .static_config + .revert_config + .revert_up_to_and_including = revert_up_to_and_including; + config + .consensus_manager_config + .as_mut() + .unwrap() + .revert_config + .revert_up_to_and_including = revert_up_to_and_including; + } +} + +fn modify_height_configs_idle_nodes( + integration_test_manager: &mut IntegrationTestManager, + node_indices: HashSet, +) { + integration_test_manager.modify_config_idle_nodes(node_indices, |_config| { + // TODO(noamsp): Change these values point to a single config value and refactor this + // function accordingly. + }); +} diff --git a/crates/apollo_integration_tests/src/flows/sync.rs b/crates/apollo_integration_tests/src/flows/sync.rs new file mode 100644 index 00000000000..337c65f5b48 --- /dev/null +++ b/crates/apollo_integration_tests/src/flows/sync.rs @@ -0,0 +1,60 @@ +use apollo_infra_utils::test_utils::TestIdentifier; +use apollo_node_config::component_execution_config::{ + ActiveComponentExecutionMode, + ReactiveComponentExecutionMode, +}; +use apollo_node_config::node_config::SequencerNodeConfig; +use apollo_state_sync_config::config::CentralSyncClientConfig; +use starknet_api::block::BlockNumber; +use tracing::info; + +use crate::integration_test_manager::IntegrationTestManager; +use crate::integration_test_utils::integration_test_setup; +use crate::utils::NodeDescriptor; + +pub async fn run() { + integration_test_setup("sync").await; + const BLOCK_TO_WAIT_FOR: BlockNumber = BlockNumber(20); + // Node layout (index → type): 0 = consolidated, 1 = consolidated. + // Node 1 is configured as the central sync node. + const CENTRAL_SYNC_NODE: usize = 1; + + let node_descriptors = vec![NodeDescriptor::consolidated(), NodeDescriptor::consolidated()]; + + let mut integration_test_manager = IntegrationTestManager::new( + node_descriptors, + None, + TestIdentifier::SyncFlowIntegrationTest, + ) + .await; + + let update_config_disable_everything_but_sync = |config: &mut SequencerNodeConfig| { + config.components.batcher.execution_mode = ReactiveComponentExecutionMode::Disabled; + config.components.gateway.execution_mode = ReactiveComponentExecutionMode::Disabled; + config.components.mempool.execution_mode = ReactiveComponentExecutionMode::Disabled; + config.components.mempool_p2p.execution_mode = ReactiveComponentExecutionMode::Disabled; + config.components.l1_events_provider.execution_mode = + ReactiveComponentExecutionMode::Disabled; + config.components.consensus_manager.execution_mode = ActiveComponentExecutionMode::Disabled; + config.components.http_server.execution_mode = ActiveComponentExecutionMode::Disabled; + config.components.l1_events_scraper.execution_mode = ActiveComponentExecutionMode::Disabled; + }; + + let update_config_use_central_sync = |config: &mut SequencerNodeConfig| { + config.state_sync_config.as_mut().unwrap().static_config.central_sync_client_config = + Some(CentralSyncClientConfig::default()); + config.state_sync_config.as_mut().unwrap().static_config.p2p_sync_client_config = None; + }; + + let node_indices = integration_test_manager.get_node_indices(); + integration_test_manager + .modify_config_idle_nodes(node_indices.clone(), update_config_disable_everything_but_sync); + integration_test_manager + .modify_config_idle_nodes([CENTRAL_SYNC_NODE].into(), update_config_use_central_sync); + + integration_test_manager.run_nodes(node_indices.clone()).await; + + integration_test_manager.await_sync_block_on_all_running_nodes(BLOCK_TO_WAIT_FOR).await; + + info!("Sync flow integration test completed successfully!"); +} diff --git a/crates/apollo_integration_tests/src/lib.rs b/crates/apollo_integration_tests/src/lib.rs index 92e8dd84bf8..f0c6fe00d30 100644 --- a/crates/apollo_integration_tests/src/lib.rs +++ b/crates/apollo_integration_tests/src/lib.rs @@ -1,5 +1,6 @@ pub mod executable_setup; pub mod flow_test_setup; +pub mod flows; pub mod integration_test_manager; pub mod integration_test_utils; pub mod monitoring_utils;