feat(starknet_integration_test): add await revert all running nodes fn

crates/starknet_integration_tests/src/sequencer_manager.rs

@@ -1,6 +1,7 @@
 use std::collections::{HashMap, HashSet};
 use std::net::SocketAddr;
 use std::path::PathBuf;
+use std::time::Duration;
 
 use futures::future::join_all;
 use futures::TryFutureExt;
@@ -16,13 +17,15 @@ use starknet_infra_utils::test_utils::{
     TestIdentifier,
     MAX_NUMBER_OF_INSTANCES_PER_TEST,
 };
+use starknet_infra_utils::tracing::{CustomLogger, TraceLevel};
 use starknet_monitoring_endpoint::test_utils::MonitoringClient;
 use starknet_sequencer_node::config::component_config::ComponentConfig;
 use starknet_sequencer_node::config::component_execution_config::{
     ActiveComponentExecutionConfig,
     ReactiveComponentExecutionConfig,
 };
 use starknet_sequencer_node::test_utils::node_runner::{get_node_executable_path, spawn_run_node};
+use tokio::join;
 use tokio::task::JoinHandle;
 use tracing::info;
 
@@ -40,7 +43,7 @@ use crate::utils::{
     TestScenario,
 };
 const DEFAULT_SENDER_ACCOUNT: AccountId = 0;
-const BLOCK_TO_WAIT_FOR_BOOTSTRAP: BlockNumber = BlockNumber(2);
+pub const BLOCK_TO_WAIT_FOR_BOOTSTRAP: BlockNumber = BlockNumber(2);
 
 /// Holds the component configs for a set of sequencers, composing a single sequencer node.
 pub struct NodeComponentConfigs {
@@ -250,6 +253,69 @@ impl IntegrationTestManager {
         });
     }
 
+    pub async fn await_revert_all_running_nodes(
+        &self,
+        expected_block_marker: BlockNumber,
+        timeout_duration: Duration,
+        interval_ms: u64,
+        max_attempts: usize,
+    ) {
+        info!("Waiting for all idle nodes to finish reverting.");
+        let condition =
+            |&latest_block_number: &BlockNumber| latest_block_number == expected_block_marker;
+
+        let await_reverted_tasks = self.running_nodes.values().map(|running_node| async {
+            let running_node_setup = &running_node.node_setup;
+            let batcher_logger = CustomLogger::new(
+                TraceLevel::Info,
+                Some(format!(
+                    "Waiting for batcher height metric to reach block {expected_block_marker} in \
+                     sequencer {} executable {}.",
+                    running_node_setup.get_node_index().unwrap(),
+                    running_node_setup.get_batcher_index(),
+                )),
+            );
+
+            // TODO(noamsp): rename batcher index/monitoringClient or use sync
+            // index/monitoringClient
+            let sync_logger = CustomLogger::new(
+                TraceLevel::Info,
+                Some(format!(
+                    "Waiting for sync height metric to reach block {expected_block_marker} in \
+                     sequencer {} executable {}.",
+                    running_node_setup.get_node_index().unwrap(),
+                    running_node_setup.get_batcher_index(),
+                )),
+            );
+
+            join!(
+                monitoring_utils::await_batcher_block(
+                    interval_ms,
+                    condition,
+                    max_attempts,
+                    running_node_setup.batcher_monitoring_client(),
+                    batcher_logger,
+                ),
+                monitoring_utils::await_sync_block_marker(
+                    interval_ms,
+                    condition,
+                    max_attempts,
+                    running_node_setup.batcher_monitoring_client(),
+                    sync_logger,
+                )
+            )
+        });
+
+        tokio::time::timeout(timeout_duration, join_all(await_reverted_tasks))
+            .await
+            .expect("Running Nodes should be reverted.");
+
+        info!(
+            "All running nodes have been reverted succesfully to block marker \
+             {expected_block_marker}."
+        );
+    }
+
     pub fn get_node_indices(&self) -> HashSet<usize> {
         self.node_indices.clone()
     }