refactor(poa): move redis publish into PoA, native-async with cached connections

.changes/changed/3280.md

@@ -0,0 +1 @@
+Move the PoA redis-publish path out of the importer and into the PoA service itself. `BlockReconciliationWritePort` now lives in `fuel_core_poa::ports` and is async; `RedisLeaderLeaseAdapter::publish_produced_block` runs natively on the tokio runtime, fans out to all nodes in parallel via `FuturesUnordered`, reuses cached per-node `MultiplexedConnection`s, wraps each per-node call in `tokio::time::timeout(node_timeout, ...)`, and short-circuits on quorum by dropping the remaining futures. No more rayon bridge, no more `std::thread::spawn` leak from the 2026-04-22 hotfix.

crates/fuel-core/src/service/adapters/block_importer.rs

@@ -7,7 +7,6 @@ use crate::{
         BlockImporterAdapter,
         ExecutorAdapter,
         VerifierAdapter,
-        consensus_module::poa::RedisLeaderLeaseAdapter,
     },
 };
 use fuel_core_importer::{
@@ -62,30 +61,16 @@ use fuel_core_types::{
 use itertools::Itertools;
 use std::sync::Arc;
 
-#[allow(clippy::large_enum_variant)]
-pub enum BlockReconciliationWriteAdapter {
-    Redis(RedisLeaderLeaseAdapter),
-    Noop(NoopBlockReconciliationWriteAdapter),
-}
-
 impl BlockImporterAdapter {
     pub fn new(
         chain_id: ChainId,
         config: Config,
         database: Database,
         executor: ExecutorAdapter,
         verifier: VerifierAdapter,
-        block_reconciliation_write_adapter: BlockReconciliationWriteAdapter,
     ) -> Self {
         let database_for_height = database.clone();
-        let importer = Importer::new(
-            chain_id,
-            config,
-            database,
-            executor,
-            verifier,
-            block_reconciliation_write_adapter,
-        );
+        let importer = Importer::new(chain_id, config, database, executor, verifier);
         Self {
             block_importer: Arc::new(importer),
             database: database_for_height,
@@ -111,32 +96,6 @@ impl BlockVerifier for VerifierAdapter {
     }
 }
 
-#[derive(Default)]
-pub struct NoopBlockReconciliationWriteAdapter;
-
-impl fuel_core_importer::ports::BlockReconciliationWritePort
-    for BlockReconciliationWriteAdapter
-{
-    fn publish_produced_block(&self, block: &SealedBlock) -> anyhow::Result<()> {
-        match self {
-            Self::Redis(adapter) => {
-                fuel_core_importer::ports::BlockReconciliationWritePort::publish_produced_block(adapter, block)
-            }
-            Self::Noop(adapter) => {
-                fuel_core_importer::ports::BlockReconciliationWritePort::publish_produced_block(adapter, block)
-            }
-        }
-    }
-}
-
-impl fuel_core_importer::ports::BlockReconciliationWritePort
-    for NoopBlockReconciliationWriteAdapter
-{
-    fn publish_produced_block(&self, _block: &SealedBlock) -> anyhow::Result<()> {
-        Ok(())
-    }
-}
-
 impl ImporterDatabase for Database {
     fn latest_block_height(&self) -> StorageResult<Option<BlockHeight>> {
         self.iter_all_keys::<FuelBlocks>(Some(IterDirection::Reverse))