apollo_integration_tests: add restart service on single node test (#10456)

Author: noamsp-starkware

crates/apollo_infra_utils/src/test_utils.rs

@@ -36,6 +36,7 @@ pub enum TestIdentifier {
     PositiveFlowIntegrationTest,
     RestartFlowIntegrationTest,
     RestartServiceMultipleNodesFlowIntegrationTest,
+    RestartServiceSingleNodeFlowIntegrationTest,
     RevertFlowIntegrationTest,
     SystemTestDumpSingleNodeConfig,
     HttpServerUnitTests,

crates/apollo_integration_tests/Cargo.toml

@@ -100,6 +100,10 @@ path = "src/bin/sequencer_node_end_to_end_integration_tests/integration_test_res
 name = "integration_test_restart_service_multiple_nodes_flow"
 path = "src/bin/sequencer_node_end_to_end_integration_tests/integration_test_restart_service_multiple_nodes_flow.rs"
 
+[[bin]]
+name = "integration_test_restart_service_single_node_flow"
+path = "src/bin/sequencer_node_end_to_end_integration_tests/integration_test_restart_service_single_node_flow.rs"
+
 [[bin]]
 name = "integration_test_revert_flow"
 path = "src/bin/sequencer_node_end_to_end_integration_tests/integration_test_revert_flow.rs"

crates/apollo_integration_tests/src/bin/sequencer_node_end_to_end_integration_tests/integration_test_restart_flow.rs

@@ -9,7 +9,6 @@ use apollo_integration_tests::integration_test_manager::{
 };
 use apollo_integration_tests::integration_test_utils::integration_test_setup;
 use strum::IntoEnumIterator;
-use tokio::select;
 use tracing::info;
 
 #[tokio::main]
@@ -74,11 +73,6 @@ async fn main() {
     // Create a simulator for sustained transaction sending.
     let simulator = integration_test_manager.create_simulator();
     let mut tx_generator = integration_test_manager.tx_generator().snapshot();
-    // TODO(noamsp/itay): Try to refactor this test to spawn threads for each task instead of using
-    // select!
-    // Task that sends sustained transactions without ever finishing.
-    let continuous_tx_sending_task =
-        simulator.run_simulation(&mut tx_generator, true, DEFAULT_SENDER_ACCOUNT);
 
     // Task that awaits transactions and restarts nodes in phases.
     let await_and_restart_nodes_task = async {
@@ -116,15 +110,13 @@ async fn main() {
         integration_test_manager.poll_all_running_nodes_received_more_txs(LONG_TIMEOUT).await;
     };
 
-    select! {
-        _ = continuous_tx_sending_task => {
-            panic!("This task should never complete");
-        }
-        _ = await_and_restart_nodes_task => {
-            // If await_and_restart_nodes_task completes normally, it finished successfully.
-            // If it panicked, the panic would have already propagated.
-        }
-    }
+    simulator
+        .run_test_with_nonstop_tx_sending(
+            &mut tx_generator,
+            DEFAULT_SENDER_ACCOUNT,
+            await_and_restart_nodes_task,
+        )
+        .await;
 
     integration_test_manager.shutdown_nodes(node_indices);
     info!("Restart flow integration test completed successfully!");

crates/apollo_integration_tests/src/bin/sequencer_node_end_to_end_integration_tests/integration_test_restart_service_single_node_flow.rs

@@ -0,0 +1,104 @@
+use std::collections::HashMap;
+use std::time::Duration;
+
+use apollo_deployments::deployments::hybrid::HybridNodeServiceName;
+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 strum::IntoEnumIterator;
+use tracing::info;
+
+#[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);
+    /// The number of consolidated local sequencers that participate in the test.
+    const N_CONSOLIDATED_SEQUENCERS: usize = 0;
+    /// The number of distributed remote sequencers that participate in the test.
+    const N_DISTRIBUTED_SEQUENCERS: usize = 0;
+    /// The number of hybrid sequencers that participate in the test.
+    const N_HYBRID_SEQUENCERS: usize = 3;
+    // The indices of the nodes that we will be shutting down.
+    // The test restarts a hybrid node's service and shuts down a non-consolidated
+    // (hybrid/distributed) node.
+    const RESTART_NODE: usize = N_HYBRID_SEQUENCERS - 1;
+
+    // Get the sequencer configurations.
+    let mut integration_test_manager = IntegrationTestManager::new(
+        N_CONSOLIDATED_SEQUENCERS,
+        N_DISTRIBUTED_SEQUENCERS,
+        N_HYBRID_SEQUENCERS,
+        None,
+        TestIdentifier::RestartServiceSingleNodeFlowIntegrationTest,
+    )
+    .await;
+
+    // Assert that RESTART_NODE is a hybrid node.
+    assert_eq!(
+        integration_test_manager
+            .get_idle_nodes()
+            .get(&RESTART_NODE)
+            .unwrap()
+            .get_executables()
+            .len(),
+        HybridNodeServiceName::iter().count()
+    );
+
+    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.shutdown_nodes(node_indices);
+    info!("Restart service single node flow integration test completed successfully!");
+}

crates/apollo_integration_tests/src/sequencer_simulator_utils.rs

@@ -1,3 +1,4 @@
+use std::future::Future;
 use std::net::{SocketAddr, ToSocketAddrs};
 use std::time::Duration;
 
@@ -6,6 +7,7 @@ use apollo_monitoring_endpoint::test_utils::MonitoringClient;
 use mempool_test_utils::starknet_api_test_utils::{AccountId, MultiAccountTransactionGenerator};
 use starknet_api::rpc_transaction::RpcTransaction;
 use starknet_api::transaction::TransactionHash;
+use tokio::select;
 use tokio::time::sleep;
 use tracing::info;
 use url::Url;
@@ -117,6 +119,26 @@ impl SequencerSimulator {
             i += 1;
         }
     }
+
+    pub async fn run_test_with_nonstop_tx_sending(
+        &self,
+        tx_generator: &mut MultiAccountTransactionGenerator,
+        sender_account: AccountId,
+        test_future: impl Future<Output = ()>,
+    ) {
+        // TODO(noamsp/itay): Try to refactor this to spawn threads for each task instead of using
+        // select!
+        let simulation_task = self.run_simulation(tx_generator, true, sender_account);
+        select! {
+            _ = simulation_task => {
+                panic!("Simulation task should not complete before the test task");
+            }
+            _ = test_future => {
+                // If test_future completes normally, it finished successfully.
+                // If it panicked, the panic would have already propagated.
+            }
+        }
+    }
 }
 
 fn get_socket_addr(url_str: &str, port: u16) -> Result<SocketAddr, Box<dyn std::error::Error>> {

scripts/run_tests.py

@@ -24,6 +24,7 @@
     "integration_test_positive_flow",
     "integration_test_restart_flow",
     "integration_test_restart_service_multiple_nodes_flow",
+    "integration_test_restart_service_single_node_flow",
     "integration_test_revert_flow",
     "integration_test_central_and_p2p_sync_flow",
 ]

scripts/sequencer_integration_test.sh

@@ -9,6 +9,7 @@
 #   - "positive" to run the positive flow test,
 #   - "restart" to run the restart flow test,
 #   - "restart_multiple_nodes" to run the restart flow test with multiple nodes,
+#   - "restart_single_node" to run the restart flow test with a single node,
 #   - "revert" to run the revert flow test,
 #   - "sync" to run the central and p2p sync flow test,
 #   - "all" to run all tests.
@@ -22,6 +23,7 @@ declare -A TEST_ALIASES=(
   [positive]="integration_test_positive_flow"
   [restart]="integration_test_restart_flow"
   [restart_multiple_nodes]="integration_test_restart_service_multiple_nodes_flow"
+  [restart_single_node]="integration_test_restart_service_single_node_flow"
   [revert]="integration_test_revert_flow"
   [sync]="integration_test_central_and_p2p_sync_flow"
 )