test(consensus): fix foreign_block_distribution test

.github/workflows/ci.yml

@@ -26,7 +26,7 @@ env:
 jobs:
   fmt:
     name: fmt
-    runs-on: [self-hosted, ubuntu-high-cpu]
+    runs-on: [ self-hosted, ubuntu-high-cpu ]
 
     steps:
       - name: checkout
@@ -50,7 +50,7 @@ jobs:
         run: cargo +${{ env.nightly_toolchain }} fmt --all -- --check
   prettier:
     name: prettier
-    runs-on: [self-hosted, ubuntu-high-cpu]
+    runs-on: [ self-hosted, ubuntu-high-cpu ]
 
     steps:
       - name: checkout
@@ -66,7 +66,7 @@ jobs:
 
   clippy:
     name: clippy
-    runs-on: [self-hosted, ubuntu-high-cpu]
+    runs-on: [ self-hosted, ubuntu-high-cpu ]
 
     steps:
       - name: checkout
@@ -121,7 +121,7 @@ jobs:
 
   build:
     name: check nightly
-    runs-on: [self-hosted, ubuntu-high-cpu]
+    runs-on: [ self-hosted, ubuntu-high-cpu ]
 
     steps:
       - name: checkout
@@ -147,7 +147,7 @@ jobs:
 
   build-stable:
     name: check stable
-    runs-on: [self-hosted, ubuntu-high-cpu]
+    runs-on: [ self-hosted, ubuntu-high-cpu ]
 
     steps:
       - name: checkout
@@ -196,7 +196,7 @@ jobs:
 
   test:
     name: test
-    runs-on: [self-hosted, ubuntu-high-cpu]
+    runs-on: [ self-hosted, ubuntu-high-cpu ]
 
     steps:
       - name: checkout
@@ -226,6 +226,14 @@ jobs:
       - name: cargo test
         run: cargo nextest run --all-features --release -E "not package(integration_tests)" --profile ci
 
+      # TODO: remove
+      - name: update test db
+        uses: actions/upload-artifact@v4 # upload test results as artifact
+        if: success() || failure()
+        with:
+          name: testdbs
+          path: ${{ github.workspace }}/data/*.db
+
       - name: upload artifact
         uses: actions/upload-artifact@v4 # upload test results as artifact
         if: success() || failure()

dan_layer/consensus/src/hotstuff/proposer.rs

@@ -55,7 +55,7 @@ where TConsensusSpec: ConsensusSpec
         }
         info!(
             target: LOG_TARGET,
-            "🌿 PROPOSING foreignly new locked block {} to {} foreign shards. justify: {} ({}), parent: {}",
+            "🌿 PROPOSING new locked block {} to {} foreign shards. justify: {} ({}), parent: {}",
             block,
             non_local_shards.len(),
             block.justify().block_id(),

dan_layer/consensus_tests/src/consensus.rs

@@ -243,7 +243,6 @@ async fn multi_shard_propose_blocks_with_new_transactions_until_all_committed()
     test.assert_clean_shutdown().await;
 }
 
-#[ignore = "FIXME: This test is very flaky"]
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 async fn foreign_shard_decides_to_abort() {
     setup_logger();
@@ -408,9 +407,17 @@ async fn leader_failure_node_goes_down() {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 async fn foreign_block_distribution() {
-    setup_logger();
+    crate::support::logging::setup_logger_ci();
+    std::fs::create_dir_all(std::path::PathBuf::from(env!["CARGO_MANIFEST_DIR"]).join("../../data")).unwrap();
+
     let mut test = Test::builder()
         .with_test_timeout(Duration::from_secs(60))
+        .debug_sql(
+            std::path::PathBuf::from(env!["CARGO_MANIFEST_DIR"])
+                .join("../../data/test{}.db")
+                .display()
+                .to_string(),
+        )
         .with_message_filter(Box::new(move |from: &TestAddress, to: &TestAddress, _| {
             match from.0.as_str() {
                 // We filter our message from each leader to the foreign committees. So we will rely on other members of
@@ -456,7 +463,6 @@ async fn foreign_block_distribution() {
     test.assert_all_validators_at_same_height().await;
 
     log::info!("total messages sent: {}", test.network().total_messages_sent());
-    log::info!("total messages filtered: {}", test.network().total_messages_filtered());
     // Each leader sends 3 proposals to the both foreign committees, so 6 messages per leader. 18 in total.
     assert_eq!(test.network().total_messages_filtered(), 18);
     test.assert_clean_shutdown().await;

dan_layer/consensus_tests/src/support/logging.rs

@@ -25,3 +25,25 @@ pub fn setup_logger() {
         // Apply globally
         .apply();
 }
+
+// TODO: tmp remove
+pub fn setup_logger_ci() {
+    let _ignore = fern::Dispatch::new()
+        // Perform allocation-free log formatting
+        .format(|out, message, record| {
+            out.finish(format_args!(
+                "{} [{}] {} {}",
+                humantime::format_rfc3339(std::time::SystemTime::now()),
+                record.target().strip_prefix("tari::dan::consensus::hotstuff::").unwrap_or(record.target()),
+                record.level(),
+                message
+            ))
+        })
+        // Add blanket level filter -
+        .level(log::LevelFilter::Debug)
+        // Output to stdout, files, and other Dispatch configurations
+        .chain(std::io::stdout())
+        // .chain(fern::log_file("output.log").unwrap())
+        // Apply globally
+        .apply();
+}

dan_layer/consensus_tests/src/support/validator/builder.rs

@@ -90,7 +90,8 @@ impl ValidatorBuilder {
         let (outbound_messaging, rx_loopback) = TestOutboundMessaging::create(tx_leader, tx_broadcast);
         let inbound_messaging = TestInboundMessaging::new(self.address.clone(), rx_hs_message, rx_loopback);
 
-        let store = SqliteStateStore::connect(&self.sql_url).unwrap();
+        let store = SqliteStateStore::connect(&self.sql_url)
+            .unwrap_or_else(|e| panic!("Failed to connect to db at {}: {}", self.sql_url, e));
         let signing_service = TestVoteSignatureService::new(self.public_key.clone(), self.address.clone());
         let transaction_pool = TransactionPool::new();
         let noop_state_manager = NoopStateManager::new();