fix(tests): flakiness across a few different tests (#2982)

> [!NOTE]
> discovered the issue, bob doesn't have the delegate keys, and
therefore when he receives preconfirmations he can't verify the
signature, and then rejects the messages. the rejections reduce the peer
score of the authority node and then the two are disconnected. maybe the
preconfirmation service should just `GossipsubMessageAcceptance::Ignore`
if it has no delegate keys instead of
`GossipsubMessageAcceptance::Reject`.

focusing on improving the configuration of nodes, enhancing consistency
checks, and cleaning up unused imports.

### Configuration Enhancements:
* Added a new configuration option
`pre_confirmation_signature_service.echo_delegation_interval` to set an
interval of 100 milliseconds for echo delegation in `make_nodes`. This
improves configurability for testing scenarios.
* Introduced conditional database initialization in `make_config`, using
`CombinedDatabase` when the `default` feature is enabled, and falling
back to an in-memory database otherwise. This provides flexibility for
different environments.

### Consistency Check Improvements:
* Refactored the `consistency` method to use a `tokio::time::interval`
for periodic checks, replacing the previous loop and event-based logic.
This simplifies the code and ensures consistent behavior.
* Consolidated consistency methods by introducing
`consistency_with_duration`, which allows specifying a custom timeout
duration. The existing `consistency_10s` and `consistency_20s` methods
now delegate to this new method for better code reuse.

### Cleanup:
* Removed an unused import `schema::tx::types::TransactionStatus` to
tidy up the codebase.

Author: rymnc

crates/fuel-core/src/p2p_test_helpers.rs

@@ -12,7 +12,6 @@ use crate::{
     },
     fuel_core_graphql_api::storage::transactions::TransactionStatuses,
     p2p::Multiaddr,
-    schema::tx::types::TransactionStatus,
     service::{
         Config,
         FuelService,
@@ -401,6 +400,12 @@ pub async fn make_nodes(
 
             node_config.consensus_signer = SignMode::Key(Secret::new(secret.into()));
 
+            if !txs.is_empty() {
+                node_config
+                    .pre_confirmation_signature_service
+                    .echo_delegation_interval = Duration::from_millis(200);
+            }
+
             test_txs = txs;
         }
 
@@ -507,7 +512,14 @@ pub fn make_config(
 }
 
 pub async fn make_node(node_config: Config, test_txs: Vec<Transaction>) -> Node {
+    #[cfg(feature = "default")]
+    let db = CombinedDatabase::from_config(&node_config.combined_db_config)
+        .unwrap()
+        .on_chain()
+        .clone();
+    #[cfg(not(feature = "default"))]
     let db = Database::in_memory();
+
     // Test coverage slows down the execution a lot, and while running all tests,
     // it may require a lot of time to start the node. We have a
     // timeout here to watch infinity loops, so it is okay to use 120 seconds.
@@ -573,55 +585,55 @@ impl Node {
     /// Wait for the node to reach consistency with the given transactions.
     pub async fn consistency(&mut self, txs: &HashMap<Bytes32, Transaction>) {
         let db = self.node.shared.database.off_chain();
-        loop {
-            let not_found = not_found_txs(db, txs);
-
-            if not_found.is_empty() {
-                break;
-            }
-
-            let tx_id = not_found[0];
-            let mut wait_transaction =
-                self.node.transaction_status_change(tx_id).await.unwrap();
+        let mut interval = tokio::time::interval(tokio::time::Duration::from_millis(100));
 
-            loop {
-                tokio::select! {
-                    result = wait_transaction.next() => {
-                        let status = result.unwrap().unwrap();
+        // first tick completes immediately
+        interval.tick().await;
 
-                        if matches!(status, TransactionStatus::Failure { .. })
-                            || matches!(status, TransactionStatus::Success { .. }) {
-                            break
-                        }
+        loop {
+            tokio::select! {
+                _ = interval.tick() => {
+                    let not_found = not_found_txs(db, txs);
 
-                        if matches!(status, TransactionStatus::SqueezedOut(_)) {
-                            panic!("Transaction was squeezed out: {:?}", status);
-                        }
-                    }
-                    _ = self.node.await_shutdown() => {
-                        panic!("Got a stop signal")
+                    if not_found.is_empty() {
+                        break;
                     }
                 }
+                _ = self.node.await_shutdown() => {
+                    panic!("Got a stop signal")
+                }
             }
         }
     }
 
-    /// Wait for the node to reach consistency with the given transactions within 10 seconds.
-    pub async fn consistency_10s(&mut self, txs: &HashMap<Bytes32, Transaction>) {
-        tokio::time::timeout(Duration::from_secs(10), self.consistency(txs))
+    pub async fn consistency_with_duration(
+        &mut self,
+        txs: &HashMap<Bytes32, Transaction>,
+        duration: Duration,
+    ) {
+        tokio::time::timeout(duration, self.consistency(txs))
             .await
             .unwrap_or_else(|_| {
                 panic!("Failed to reach consistency for {:?}", self.config.name)
             });
     }
 
+    /// Wait for the node to reach consistency with the given transactions within 10 seconds.
+    pub async fn consistency_10s(&mut self, txs: &HashMap<Bytes32, Transaction>) {
+        self.consistency_with_duration(txs, Duration::from_secs(10))
+            .await;
+    }
+
     /// Wait for the node to reach consistency with the given transactions within 20 seconds.
     pub async fn consistency_20s(&mut self, txs: &HashMap<Bytes32, Transaction>) {
-        tokio::time::timeout(Duration::from_secs(20), self.consistency(txs))
-            .await
-            .unwrap_or_else(|_| {
-                panic!("Failed to reach consistency for {:?}", self.config.name)
-            });
+        self.consistency_with_duration(txs, Duration::from_secs(20))
+            .await;
+    }
+
+    /// Wait for the node to reach consistency with the given transactions within 30 seconds.
+    pub async fn consistency_30s(&mut self, txs: &HashMap<Bytes32, Transaction>) {
+        self.consistency_with_duration(txs, Duration::from_secs(30))
+            .await;
     }
 
     /// Insert the test transactions into the node's transaction pool.

tests/tests/peering.rs

@@ -57,7 +57,7 @@ async fn max_discovery_peers_connected__node_will_not_discover_new_nodes_if_full
     assert!(actual <= expected);
 }
 
-#[tokio::test]
+#[tokio::test(flavor = "multi_thread")]
 async fn max_functional_peers_connected__nodes_will_discover_new_peers_if_first_peer_is_full()
  {
     let mut rng = StdRng::seed_from_u64(1234);
@@ -87,7 +87,7 @@ async fn max_functional_peers_connected__nodes_will_discover_new_peers_if_first_
         None,
     )
     .await;
-    tokio::time::sleep(Duration::from_secs(1)).await;
+    tokio::time::sleep(Duration::from_secs(10)).await;
 
     // then
     let Nodes {
@@ -98,7 +98,7 @@ async fn max_functional_peers_connected__nodes_will_discover_new_peers_if_first_
     let producer = producers.pop().unwrap();
     let client = FuelClient::from(producer.node.bound_address);
     client.produce_blocks(new_blocks, None).await.unwrap();
-    tokio::time::sleep(Duration::from_secs(10)).await;
+    tokio::time::sleep(Duration::from_secs(20)).await;
 
     for validator in validators {
         let client = FuelClient::from(validator.node.bound_address);

tests/tests/sync.rs

@@ -294,7 +294,7 @@ async fn test_multiple_producers_different_keys() {
     for (expected, mut validators) in expected.iter().zip(groups) {
         assert_eq!(group_size, validators.len());
         for v in &mut validators {
-            v.consistency_20s(expected).await;
+            v.consistency_30s(expected).await;
         }
     }
 }