Fix flaky consensus smoke test

.github/workflows/ci.yml

@@ -56,7 +56,7 @@ jobs:
       - name: Compile unit tests with all features enabled
         run: cargo nextest run --cargo-profile ci-dev --all-targets --all-features --workspace --locked --no-run --timings
       - name: Run unit tests with all features enabled
-        run: timeout 10m cargo nextest run --cargo-profile ci-dev --no-fail-fast --all-targets --all-features --workspace --locked
+        run: PATHFINDER_CONSENSUS_TEST_DUMP_CHILD_LOGS_ON_FAIL=1 timeout 10m cargo nextest run --cargo-profile ci-dev --no-fail-fast --all-targets --all-features --workspace --locked
       - name: Store timings with all features enabled
         uses: actions/upload-artifact@v4
         with:

crates/consensus/src/lib.rs

@@ -831,7 +831,7 @@ impl<A: Clone + Ord> ValidatorSet<A> {
             .map(|v| (v.address.clone(), v))
             .collect();
         assert!(!validators.is_empty());
-        let validators = validators.into_values().collect();
+        let validators = validators.into_values().collect::<Vec<Validator<A>>>();
         Self { validators }
     }
 

crates/executor/src/state_reader/storage_adapter/concurrent.rs

@@ -67,6 +67,7 @@ enum Command {
         ClassHash,
         SyncSender<Result<Option<CasmHash>, StateError>>,
     ),
+    Exit,
 }
 
 impl ConcurrentStorageAdapter {
@@ -79,6 +80,12 @@ impl ConcurrentStorageAdapter {
     }
 }
 
+impl Drop for ConcurrentStorageAdapter {
+    fn drop(&mut self) {
+        _ = self.tx.send(Command::Exit);
+    }
+}
+
 impl StorageAdapter for ConcurrentStorageAdapter {
     fn transaction_executor_config(&self) -> TransactionExecutorConfig {
         let n_workers = std::thread::available_parallelism()
@@ -248,6 +255,7 @@ fn db_thread(
         };
 
         match command {
+            Command::Exit => return,
             Command::BlockHash(block_id, sender) => {
                 sender
                     .send(db_tx.block_hash(block_id))

crates/pathfinder/src/bin/pathfinder/main.rs

@@ -311,6 +311,7 @@ Hint: This is usually caused by exceeding the file descriptor limit of your syst
                 wal_directory,
                 client,
                 event_rx,
+                &config.data_directory,
             )
         } else {
             ConsensusTaskHandles::pending()

crates/pathfinder/src/config.rs

@@ -595,7 +595,7 @@ struct NativeExecutionCli {
         default_value = "false",
         env = "PATHFINDER_RPC_NATIVE_EXECUTION"
     )]
-    is_enabled: bool,
+    is_native_execution_enabled: bool,
 
     #[arg(
         long = "rpc.native-execution-class-cache-size",
@@ -617,15 +617,15 @@ struct ConsensusCli {
         default_value = "false",
         env = "PATHFINDER_CONSENSUS_ENABLE",
     )]
-    is_enabled: bool,
+    is_consensus_enabled: bool,
 
     #[arg(
         long = "consensus.proposer-address",
         long_help = "Address of the sole proposer for the consensus protocol.",
         value_name = "ADDRESS",
         value_parser = parse_felt,
         env = "PATHFINDER_CONSENSUS_PROPOSER_ADDRESS",
-        required_if_eq("is_enabled", "true"),
+        required_if_eq("is_consensus_enabled", "true"),
     )]
     proposer_address: Option<Felt>,
 
@@ -635,7 +635,7 @@ struct ConsensusCli {
         value_name = "ADDRESS",
         value_parser = parse_felt,
         env = "PATHFINDER_CONSENSUS_MY_VALIDATOR_ADDRESS",
-        required_if_eq("is_enabled", "true"),
+        required_if_eq("is_consensus_enabled", "true"),
     )]
     my_validator_address: Option<Felt>,
 
@@ -646,7 +646,7 @@ struct ConsensusCli {
         value_parser = parse_felt,
         value_delimiter = ',',
         env = "PATHFINDER_CONSENSUS_VALIDATOR_ADDRESSES",
-        required_if_eq("is_enabled", "true"),
+        required_if_eq("is_consensus_enabled", "true"),
     )]
     validator_addresses: Vec<Felt>,
 }
@@ -977,7 +977,7 @@ impl NativeExecutionConfig {
 impl NativeExecutionConfig {
     fn parse(args: NativeExecutionCli) -> Self {
         Self {
-            enabled: args.is_enabled,
+            enabled: args.is_native_execution_enabled,
             class_cache_size: args.class_cache_size,
         }
     }
@@ -1001,17 +1001,16 @@ impl ConsensusConfig {
 #[cfg(feature = "p2p")]
 impl ConsensusConfig {
     fn parse_or_exit(args: ConsensusCli) -> Option<Self> {
-        args.is_enabled.then(|| {
-            if std::iter::once(
-                &args
-                    .my_validator_address
-                    .expect("Required if `is_enabled` is true"),
-            )
-            .chain(args.validator_addresses.iter())
-            .collect::<HashSet<_>>()
-            .len()
-                < 3
-            {
+        args.is_consensus_enabled.then(|| {
+            let my_validator_address = args
+                .my_validator_address
+                .as_ref()
+                .expect("Required if `is_consensus_enabled` is true");
+            let unique_validator_addresses = std::iter::once(my_validator_address)
+                .chain(args.validator_addresses.iter())
+                .collect::<HashSet<_>>();
+
+            if unique_validator_addresses.len() < 3 {
                 Cli::command()
                     .error(
                         clap::error::ErrorKind::ValueValidation,
@@ -1025,11 +1024,11 @@ impl ConsensusConfig {
             Self {
                 proposer_address: ContractAddress(
                     args.proposer_address
-                        .expect("Required if `is_enabled` is true"),
+                        .expect("Required if `is_consensus_enabled` is true"),
                 ),
                 my_validator_address: ContractAddress(
                     args.my_validator_address
-                        .expect("Required if `is_enabled` is true"),
+                        .expect("Required if `is_consensus_enabled` is true"),
                 ),
                 validator_addresses: args
                     .validator_addresses