reth_network/transactions/

mod.rs

//! Transactions management for the p2p network.

use alloy_consensus::{constants::EIP4844_TX_TYPE_ID, transaction::TxHashRef};
use rayon::iter::{IntoParallelIterator, ParallelIterator};
use smallvec::SmallVec;

/// Normalized transaction announcements.
pub mod announcement;

/// Aggregation on configurable parameters for [`TransactionsManager`].
pub mod config;
/// Default and spec'd bounds.
pub mod constants;
/// Component responsible for fetching transactions from [`NewPooledTransactionHashes`].
pub mod fetcher;
/// Defines the traits for transaction-related policies.
pub mod policy;
#[cfg(test)]
mod test_harness;

pub use self::constants::{
    tx_fetcher::DEFAULT_SOFT_LIMIT_BYTE_SIZE_POOLED_TRANSACTIONS_RESP_ON_PACK_GET_POOLED_TRANSACTIONS_REQ,
    SOFT_LIMIT_BYTE_SIZE_POOLED_TRANSACTIONS_RESPONSE,
};
use announcement::TransactionAnnouncement;
use config::AnnouncementAcceptance;
pub use config::{
    AnnouncementFilteringPolicy, TransactionFetcherConfig, TransactionIngressPolicy,
    TransactionPropagationMode, TransactionPropagationPolicy, TransactionsManagerConfig,
};
use policy::NetworkPolicies;

pub(crate) use fetcher::{FetchEvent, TransactionFetcher};

use self::constants::{
    tx_manager::*, DEFAULT_SOFT_LIMIT_BYTE_SIZE_TRANSACTIONS_BROADCAST_MESSAGE,
    SOFT_LIMIT_COUNT_HASHES_IN_GET_POOLED_TRANSACTIONS_REQUEST,
};
use crate::{
    budget::{
        DEFAULT_BUDGET_TRY_DRAIN_NETWORK_TRANSACTION_EVENTS,
        DEFAULT_BUDGET_TRY_DRAIN_PENDING_POOL_IMPORTS, DEFAULT_BUDGET_TRY_DRAIN_STREAM,
    },
    cache::LruCache,
    duration_metered_exec, metered_poll_nested_stream_with_budget,
    metrics::{AnnouncedTxTypesMetrics, TransactionsManagerMetrics},
    transactions::config::{StrictEthAnnouncementFilter, TransactionPropagationKind},
    NetworkHandle, TxTypesCounter,
};
use alloy_eips::eip2718::{Encodable2718, Typed2718};
use alloy_primitives::{
    bytes::BufMut,
    map::{hash_map::Entry, B256Map, B256Set, FbBuildHasher, HashMap, HashSet},
    TxHash, B256,
};
use alloy_rlp::Encodable;
use constants::SOFT_LIMIT_COUNT_HASHES_IN_NEW_POOLED_TRANSACTIONS_BROADCAST_MESSAGE;
use futures::{stream::FuturesUnordered, Future, StreamExt};
use reth_eth_wire::{
    BroadcastPoolTransactions, Cells, EthNetworkPrimitives, EthVersion, GetCells,
    GetPooledTransactions, LazyEncoded, LazyEncodedTransaction, NetworkPrimitives,
    NewPooledTransactionHashes, NewPooledTransactionHashes66, NewPooledTransactionHashes68,
    NewPooledTransactionHashes72, PooledTransactions, Transactions,
};
use reth_ethereum_primitives::TxType;
use reth_evm::SenderRecoveryCache;
use reth_metrics::common::mpsc::MemoryBoundedReceiver;
use reth_network_api::{
    events::{PeerEvent, SessionInfo},
    NetworkEvent, NetworkEventListenerProvider, PeerKind, PeerRequest, PeerRequestSender, Peers,
};
use reth_network_p2p::{
    error::{RequestError, RequestResult},
    sync::SyncStateProvider,
};
use reth_network_peers::PeerId;
use reth_network_types::ReputationChangeKind;
use reth_primitives_traits::{InMemorySize, SignedTransaction};
use reth_tokio_util::EventStream;
use reth_transaction_pool::{
    error::{PoolError, PoolResult},
    AddedTransactionOutcome, GetPooledTransactionLimit, PoolTransaction, PropagateKind,
    PropagatedTransactions, TransactionPool, ValidPoolTransaction,
};
use std::{
    pin::Pin,
    sync::{
        atomic::{AtomicUsize, Ordering},
        Arc,
    },
    task::{Context, Poll},
    time::{Duration, Instant},
};
use tokio::sync::{mpsc, oneshot, oneshot::error::RecvError};
use tokio_stream::wrappers::UnboundedReceiverStream;
use tracing::{debug, trace};

/// The future for importing transactions into the pool.
///
/// Resolves with the result of each transaction import.
pub type PoolImportFuture =
    Pin<Box<dyn Future<Output = Vec<PoolResult<AddedTransactionOutcome>>> + Send + 'static>>;

/// Api to interact with [`TransactionsManager`] task.
///
/// This can be obtained via [`TransactionsManager::handle`] and can be used to manually interact
/// with the [`TransactionsManager`] task once it is spawned.
///
/// For example [`TransactionsHandle::get_peer_transaction_hashes`] returns the transaction hashes
/// known by a specific peer.
#[derive(Debug, Clone)]
pub struct TransactionsHandle<N: NetworkPrimitives = EthNetworkPrimitives> {
    /// Command channel to the [`TransactionsManager`]
    manager_tx: mpsc::UnboundedSender<TransactionsCommand<N>>,
}

impl<N: NetworkPrimitives> TransactionsHandle<N> {
    fn send(&self, cmd: TransactionsCommand<N>) {
        let _ = self.manager_tx.send(cmd);
    }

    /// Fetch the [`PeerRequestSender`] for the given peer.
    async fn peer_handle(
        &self,
        peer_id: PeerId,
    ) -> Result<Option<PeerRequestSender<PeerRequest<N>>>, RecvError> {
        let (tx, rx) = oneshot::channel();
        self.send(TransactionsCommand::GetPeerSender { peer_id, peer_request_sender: tx });
        rx.await
    }

    /// Manually propagate the transaction that belongs to the hash.
    pub fn propagate(&self, hash: TxHash) {
        self.send(TransactionsCommand::PropagateHash(hash))
    }

    /// Manually propagate the transaction hash to a specific peer.
    ///
    /// Note: this only propagates if the pool contains the transaction.
    pub fn propagate_hash_to(&self, hash: TxHash, peer: PeerId) {
        self.propagate_hashes_to(Some(hash), peer)
    }

    /// Manually propagate the transaction hashes to a specific peer.
    ///
    /// Note: this only propagates the transactions that are known to the pool.
    pub fn propagate_hashes_to(&self, hash: impl IntoIterator<Item = TxHash>, peer: PeerId) {
        let hashes = hash.into_iter().collect::<Vec<_>>();
        if hashes.is_empty() {
            return
        }
        self.send(TransactionsCommand::PropagateHashesTo(hashes, peer))
    }

    /// Request the active peer IDs from the [`TransactionsManager`].
    pub async fn get_active_peers(&self) -> Result<HashSet<PeerId>, RecvError> {
        let (tx, rx) = oneshot::channel();
        self.send(TransactionsCommand::GetActivePeers(tx));
        rx.await
    }

    /// Manually propagate full transaction hashes to a specific peer.
    ///
    /// Do nothing if transactions are empty.
    pub fn propagate_transactions_to(&self, transactions: Vec<TxHash>, peer: PeerId) {
        if transactions.is_empty() {
            return
        }
        self.send(TransactionsCommand::PropagateTransactionsTo(transactions, peer))
    }

    /// Manually propagate the given transaction hashes to all peers.
    ///
    /// It's up to the [`TransactionsManager`] whether the transactions are sent as hashes or in
    /// full.
    pub fn propagate_transactions(&self, transactions: Vec<TxHash>) {
        if transactions.is_empty() {
            return
        }
        self.send(TransactionsCommand::PropagateTransactions(transactions))
    }

    /// Manually propagate the given transactions to all peers.
    ///
    /// It's up to the [`TransactionsManager`] whether the transactions are sent as hashes or in
    /// full.
    pub fn broadcast_transactions(
        &self,
        transactions: impl IntoIterator<Item = N::BroadcastedTransaction>,
    ) {
        let transactions =
            transactions.into_iter().map(PropagateTransaction::new).collect::<Vec<_>>();
        if transactions.is_empty() {
            return
        }
        self.send(TransactionsCommand::BroadcastTransactions(transactions))
    }

    /// Request the transaction hashes known by specific peers.
    pub async fn get_transaction_hashes(
        &self,
        peers: Vec<PeerId>,
    ) -> Result<HashMap<PeerId, B256Set>, RecvError> {
        if peers.is_empty() {
            return Ok(Default::default())
        }
        let (tx, rx) = oneshot::channel();
        self.send(TransactionsCommand::GetTransactionHashes { peers, tx });
        rx.await
    }

    /// Request the transaction hashes known by a specific peer.
    pub async fn get_peer_transaction_hashes(&self, peer: PeerId) -> Result<B256Set, RecvError> {
        let res = self.get_transaction_hashes(vec![peer]).await?;
        Ok(res.into_values().next().unwrap_or_default())
    }

    /// Requests the transactions directly from the given peer.
    ///
    /// Returns `None` if the peer is not connected.
    ///
    /// **Note**: this returns the response from the peer as received.
    pub async fn get_pooled_transactions_from(
        &self,
        peer_id: PeerId,
        hashes: Vec<B256>,
    ) -> Result<Option<Vec<N::PooledTransaction>>, RequestError> {
        let Some(peer) = self.peer_handle(peer_id).await? else { return Ok(None) };

        let (tx, rx) = oneshot::channel();
        let request = PeerRequest::GetPooledTransactions { request: hashes.into(), response: tx };
        peer.try_send(request).ok();

        rx.await?.map(|res| Some(res.0))
    }

    /// Requests cells from a specific eth/72 peer.
    ///
    /// This is the transport-level client primitive for EIP-8070. The caller supplies the
    /// transaction hashes and the common cell mask for the request. Announcement-driven batching,
    /// retries, cell verification, and blob reconstruction are layered above this method.
    ///
    /// Returns `None` if the peer is no longer connected.
    pub async fn get_cells_from(
        &self,
        peer_id: PeerId,
        hashes: Vec<B256>,
        cell_mask: alloy_primitives::B128,
    ) -> Result<Option<Cells>, RequestError> {
        if hashes.is_empty() {
            return Ok(Some(Cells { cell_mask, ..Default::default() }))
        }

        let Some(peer) = self.peer_handle(peer_id).await? else { return Ok(None) };

        let (tx, rx) = oneshot::channel();
        let request =
            PeerRequest::GetCells { request: GetCells { hashes, cell_mask }, response: tx };
        peer.try_send(request).map_err(|_| RequestError::ChannelClosed)?;

        rx.await?.map(Some)
    }
}

/// Manages transactions on top of the p2p network.
///
/// This can be spawned to another task and is supposed to be run as background service.
/// [`TransactionsHandle`] can be used as frontend to programmatically send commands to it and
/// interact with it.
///
/// The [`TransactionsManager`] is responsible for:
///    - handling incoming eth messages for transactions.
///    - serving transaction requests.
///    - propagate transactions
///
/// This type communicates with the [`NetworkManager`](crate::NetworkManager) in both directions.
///   - receives incoming network messages.
///   - sends messages to dispatch (responses, propagate tx)
///
/// It is directly connected to the [`TransactionPool`] to retrieve requested transactions and
/// propagate new transactions over the network.
///
/// It can be configured with different policies for transaction propagation and announcement
/// filtering. See [`NetworkPolicies`] for more details.
///
/// ## Network Transaction Processing
///
/// ### Message Types
///
/// - **`Transactions`**: Full transaction broadcasts (rejects blob transactions)
/// - **`NewPooledTransactionHashes`**: Hash announcements
///
/// ### Peer Tracking
///
/// - Maintains per-peer transaction cache (default: 10,240 entries)
/// - Prevents duplicate imports and enables efficient propagation
///
/// ### Bad Transaction Handling
///
/// Caches and rejects transactions with consensus violations (gas, signature, chain ID).
/// Penalizes peers sending invalid transactions.
///
/// ### Import Management
///
/// Limits concurrent pool imports and backs off when approaching capacity.
///
/// ### Transaction Fetching
///
/// For announced transactions: filters known → queues unknown → fetches → imports
///
/// ### Propagation Rules
///
/// Based on: origin (Local/External/Private), peer capabilities, and network state.
/// Disabled during initial sync.
///
/// ### Security
///
/// Rate limiting via reputation, bad transaction isolation, peer scoring.
#[derive(Debug)]
#[must_use = "Manager does nothing unless polled."]
pub struct TransactionsManager<Pool, N: NetworkPrimitives = EthNetworkPrimitives> {
    /// Access to the transaction pool.
    pool: Pool,
    /// Cache of recovered transaction senders shared with payload execution, if enabled.
    sender_recovery_cache: Option<SenderRecoveryCache>,
    /// Network access.
    network: NetworkHandle<N>,
    /// Subscriptions to all network related events.
    ///
    /// From which we get all new incoming transaction related messages.
    network_events: EventStream<NetworkEvent<PeerRequest<N>>>,
    /// Transaction fetcher to handle inflight and missing transaction requests.
    transaction_fetcher: TransactionFetcher<N>,
    /// All currently pending transactions grouped by peers.
    ///
    /// This way we can track incoming transactions and prevent multiple pool imports for the same
    /// transaction
    transactions_by_peers: B256Map<SmallVec<[PeerId; 1]>>,
    /// Scratch hashes reused when deduplicating announcements.
    announcement_hashes: B256Set,
    /// Transactions that are currently imported into the `Pool`.
    ///
    /// The import process includes:
    ///  - validation of the transactions, e.g. transaction is well formed: valid tx type, fees are
    ///    valid, or for 4844 transaction the blobs are valid. See also
    ///    [`EthTransactionValidator`](reth_transaction_pool::validate::EthTransactionValidator)
    /// - if the transaction is valid, it is added into the pool.
    ///
    /// Once the new transaction reaches the __pending__ state it will be emitted by the pool via
    /// [`TransactionPool::pending_transactions_listener`] and arrive at the `pending_transactions`
    /// receiver.
    pool_imports: FuturesUnordered<PoolImportFuture>,
    /// Stats on pending pool imports that help the node self-monitor.
    pending_pool_imports_info: PendingPoolImportsInfo,
    /// Bad imports.
    bad_imports: LruCache<TxHash, FbBuildHasher<32>>,
    /// All the connected peers.
    peers: HashMap<PeerId, PeerMetadata<N>, FbBuildHasher<64>>,
    /// Send half for the command channel.
    ///
    /// This is kept so that a new [`TransactionsHandle`] can be created at any time.
    command_tx: mpsc::UnboundedSender<TransactionsCommand<N>>,
    /// Incoming commands from [`TransactionsHandle`].
    ///
    /// This will only receive commands if a user manually sends a command to the manager through
    /// the [`TransactionsHandle`] to interact with this type directly.
    command_rx: UnboundedReceiverStream<TransactionsCommand<N>>,
    /// A stream that yields new __pending__ transactions.
    ///
    /// A transaction is considered __pending__ if it is executable on the current state of the
    /// chain. In other words, this only yields transactions that satisfy all consensus
    /// requirements, these include:
    ///   - no nonce gaps
    ///   - all dynamic fee requirements are (currently) met
    ///   - account has enough balance to cover the transaction's gas
    pending_transactions: mpsc::Receiver<TxHash>,
    /// Incoming events from the [`NetworkManager`](crate::NetworkManager).
    transaction_events: MemoryBoundedReceiver<NetworkTransactionEvent<N>>,
    /// How the `TransactionsManager` is configured.
    config: TransactionsManagerConfig,
    /// Network Policies
    policies: NetworkPolicies<N>,
    /// `TransactionsManager` metrics
    metrics: TransactionsManagerMetrics,
    /// `AnnouncedTxTypes` metrics
    announced_tx_types_metrics: AnnouncedTxTypesMetrics,
}

impl<Pool: TransactionPool, N: NetworkPrimitives> TransactionsManager<Pool, N> {
    /// Sets up a new instance.
    ///
    /// Note: This expects an existing [`NetworkManager`](crate::NetworkManager) instance.
    pub fn new(
        network: NetworkHandle<N>,
        pool: Pool,
        from_network: MemoryBoundedReceiver<NetworkTransactionEvent<N>>,
        transactions_manager_config: TransactionsManagerConfig,
    ) -> Self {
        Self::with_policy(
            network,
            pool,
            from_network,
            transactions_manager_config,
            NetworkPolicies::new(
                TransactionPropagationKind::default(),
                StrictEthAnnouncementFilter::default(),
            ),
        )
    }
}

impl<Pool: TransactionPool, N: NetworkPrimitives> TransactionsManager<Pool, N> {
    /// Sets up a new instance with given the settings.
    ///
    /// Note: This expects an existing [`NetworkManager`](crate::NetworkManager) instance.
    pub fn with_policy(
        network: NetworkHandle<N>,
        pool: Pool,
        from_network: MemoryBoundedReceiver<NetworkTransactionEvent<N>>,
        transactions_manager_config: TransactionsManagerConfig,
        policies: NetworkPolicies<N>,
    ) -> Self {
        let network_events = network.event_listener();

        let (command_tx, command_rx) = mpsc::unbounded_channel();

        let transaction_fetcher =
            TransactionFetcher::new(transactions_manager_config.transaction_fetcher_config.clone());

        // install a listener for new __pending__ transactions that are allowed to be propagated
        // over the network
        let pending = pool.pending_transactions_listener();
        let pending_pool_imports_info = PendingPoolImportsInfo::new(
            transactions_manager_config.max_pending_pool_imports,
            transactions_manager_config.max_pending_pool_imports_bytes,
        );
        let metrics = TransactionsManagerMetrics::default();
        metrics
            .capacity_pending_pool_imports
            .increment(pending_pool_imports_info.max_pending_pool_imports as u64);
        metrics
            .capacity_pending_pool_imports_bytes
            .increment(pending_pool_imports_info.max_pending_pool_imports_bytes as u64);

        Self {
            pool,
            sender_recovery_cache: None,
            network,
            network_events,
            transaction_fetcher,
            transactions_by_peers: Default::default(),
            announcement_hashes: B256Set::default(),
            pool_imports: Default::default(),
            pending_pool_imports_info,
            bad_imports: LruCache::with_hasher(DEFAULT_MAX_COUNT_BAD_IMPORTS, Default::default()),
            peers: Default::default(),
            command_tx,
            command_rx: UnboundedReceiverStream::new(command_rx),
            pending_transactions: pending,
            transaction_events: from_network,
            config: transactions_manager_config,
            policies,
            metrics,
            announced_tx_types_metrics: AnnouncedTxTypesMetrics::default(),
        }
    }

    /// Returns a new handle that can send commands to this type.
    pub fn handle(&self) -> TransactionsHandle<N> {
        TransactionsHandle { manager_tx: self.command_tx.clone() }
    }

    /// Uses the provided sender recovery cache.
    pub fn with_sender_recovery_cache(mut self, cache: SenderRecoveryCache) -> Self {
        self.sender_recovery_cache = Some(cache);
        self
    }

    /// Returns whether pending pool imports are below the soft limit.
    ///
    /// Fetch responses are admitted in full when capacity remains, so one response can exceed
    /// the limit by at most 255 transactions (the 256-hash request cap minus one free slot), or
    /// the byte limit by one response message. Further responses and requests wait until imports
    /// fall below both limits again.
    fn has_capacity_for_pending_pool_imports(&self) -> bool {
        self.remaining_pool_import_capacity() > 0
    }

    /// Returns the remaining capacity below the soft limit for pending pool imports.
    ///
    /// Outstanding requests do not reserve capacity. A fetched response is admitted in full
    /// if any capacity remains; while imports exceed the count or byte limit, this returns zero.
    fn remaining_pool_import_capacity(&self) -> usize {
        let info = &self.pending_pool_imports_info;
        if info.pending_pool_imports_bytes.load(Ordering::Relaxed) >=
            info.max_pending_pool_imports_bytes
        {
            return 0
        }
        info.max_pending_pool_imports
            .saturating_sub(info.pending_pool_imports.load(Ordering::Relaxed))
    }

    /// Returns the transaction count and encoded byte size available to broadcasts, after
    /// reserving space for fetched hashes.
    fn remaining_broadcast_import_capacity(&self) -> (usize, usize) {
        let capacity = self.remaining_pool_import_capacity();
        let info = &self.pending_pool_imports_info;
        let bytes = info
            .max_pending_pool_imports_bytes
            .saturating_sub(info.pending_pool_imports_bytes.load(Ordering::Relaxed));
        if self.transaction_fetcher.num_hashes() == 0 {
            return (capacity, bytes)
        }
        // Reserve one batch (or half a small import budget) for fetched transactions.
        // Broadcasts are processed first and must not consume every newly freed slot.
        let reserved = info
            .max_pending_pool_imports
            .div_ceil(2)
            .min(SOFT_LIMIT_COUNT_HASHES_IN_GET_POOLED_TRANSACTIONS_REQUEST);
        // Likewise reserve one response worth of bytes, so broadcasts can't fill the byte limit
        // that pauses fetching.
        let reserved_bytes = info
            .max_pending_pool_imports_bytes
            .div_ceil(2)
            .min(SOFT_LIMIT_BYTE_SIZE_POOLED_TRANSACTIONS_RESPONSE);
        (capacity.saturating_sub(reserved), bytes.saturating_sub(reserved_bytes))
    }

    fn report_peer_bad_transactions(&self, peer_id: PeerId) {
        self.report_peer(peer_id, ReputationChangeKind::BadTransactions);
        self.metrics.reported_bad_transactions.increment(1);
    }

    fn report_peer(&self, peer_id: PeerId, kind: ReputationChangeKind) {
        trace!(target: "net::tx", ?peer_id, ?kind, "reporting reputation change");
        self.network.reputation_change(peer_id, kind);
    }

    fn report_already_seen(&self, peer_id: PeerId) {
        trace!(target: "net::tx", ?peer_id, "Penalizing peer for already seen transaction");
        self.network.reputation_change(peer_id, ReputationChangeKind::AlreadySeenTransaction);
    }

    /// Handles a closed peer session, removing the peer from transaction-local tracking state.
    fn on_peer_session_closed(&mut self, peer_id: &PeerId) {
        if let Some(mut peer) = self.peers.remove(peer_id) {
            self.policies.propagation_policy_mut().on_session_closed(&mut peer);
        }
        self.transaction_fetcher.on_peer_disconnected(peer_id);
    }

    /// Clear the transaction
    fn on_good_import(&mut self, hash: TxHash) {
        self.transactions_by_peers.remove(&hash);
    }

    /// Handles a failed transaction import.
    ///
    /// Blob sidecar errors (e.g. invalid proof, missing sidecar) are penalized via
    /// `report_peer_bad_transactions` but NOT cached in `bad_imports` — the transaction itself
    /// may be valid when fetched from another peer with correct sidecar data.
    ///
    /// Other bad transactions are penalized and cached in `bad_imports` to avoid fetching or
    /// importing them again.
    ///
    /// Errors that count as bad transactions are:
    ///
    /// - intrinsic gas too low
    /// - exceeds gas limit
    /// - gas uint overflow
    /// - exceeds max init code size
    /// - signer account has bytecode
    /// - chain id mismatch
    /// - old legacy chain id
    /// - tx type not supported
    ///
    /// (and additionally for blobs txns...)
    ///
    /// - no blobs
    /// - too many blobs
    /// - invalid kzg proof
    /// - kzg error
    /// - not blob transaction (tx type mismatch)
    /// - wrong versioned kzg commitment hash
    fn on_bad_import(&mut self, err: PoolError) {
        let peers = self.transactions_by_peers.remove(&err.hash);

        if err.is_bad_blob_sidecar() {
            // Blob sidecar errors: penalize but do NOT cache the hash as bad.
            // The transaction may be valid — only the sidecar from this peer was wrong.
            // Using regular penalties means repeated offenders still get disconnected.
            //
            // Eth/72 peers cannot reach this path with an elided sidecar: their blob
            // transactions are dropped before import in `import_transactions`, so any sidecar
            // failure here means actual blob data failed validation.
            if let Some(peers) = peers {
                for peer_id in peers {
                    self.report_peer_bad_transactions(peer_id);
                }
            }
            return
        }

        // if we're _currently_ syncing, we ignore a bad transaction
        if !err.is_bad_transaction() || self.network.is_syncing() {
            return
        }
        // otherwise we penalize the peer that sent the bad transaction, with the assumption that
        // the peer should have known that this transaction is bad (e.g. violating consensus rules)
        if let Some(peers) = peers {
            for peer_id in peers {
                self.report_peer_bad_transactions(peer_id);
            }
        }
        self.metrics.bad_imports.increment(1);
        self.bad_imports.insert(err.hash);
    }

    fn on_request_error(&self, peer_id: PeerId, req_err: RequestError) {
        let kind = match req_err {
            RequestError::UnsupportedCapability => ReputationChangeKind::BadProtocol,
            RequestError::Timeout => ReputationChangeKind::Timeout,
            RequestError::ChannelClosed | RequestError::ConnectionDropped => {
                // peer is already disconnected
                return
            }
            RequestError::BadResponse => return self.report_peer_bad_transactions(peer_id),
        };
        self.report_peer(peer_id, kind);
    }

    #[inline]
    fn update_poll_metrics(&self, start: Instant, poll_durations: TxManagerPollDurations) {
        let metrics = &self.metrics;

        let TxManagerPollDurations {
            acc_network_events,
            acc_pending_imports,
            acc_tx_events,
            acc_imported_txns,
            acc_fetch_events,
            acc_pending_fetch,
            acc_cmds,
        } = poll_durations;

        // update metrics for whole poll function
        metrics.duration_poll_tx_manager.set(start.elapsed().as_secs_f64());
        // update metrics for nested expressions
        metrics.acc_duration_poll_network_events.set(acc_network_events.as_secs_f64());
        metrics.acc_duration_poll_pending_pool_imports.set(acc_pending_imports.as_secs_f64());
        metrics.acc_duration_poll_transaction_events.set(acc_tx_events.as_secs_f64());
        metrics.acc_duration_poll_imported_transactions.set(acc_imported_txns.as_secs_f64());
        metrics.acc_duration_poll_fetch_events.set(acc_fetch_events.as_secs_f64());
        metrics.acc_duration_fetch_pending_hashes.set(acc_pending_fetch.as_secs_f64());
        metrics.acc_duration_poll_commands.set(acc_cmds.as_secs_f64());
    }
}

impl<Pool: TransactionPool, N: NetworkPrimitives> TransactionsManager<Pool, N> {
    /// Processes a batch import results.
    fn on_batch_import_result(&mut self, batch_results: Vec<PoolResult<AddedTransactionOutcome>>) {
        for res in batch_results {
            match res {
                Ok(AddedTransactionOutcome { hash, .. }) => {
                    self.on_good_import(hash);
                }
                Err(err) => {
                    self.on_bad_import(err);
                }
            }
        }
    }

    /// Request handler for an incoming `NewPooledTransactionHashes`
    fn on_new_pooled_transaction_hashes(
        &mut self,
        peer_id: PeerId,
        msg: NewPooledTransactionHashes,
    ) {
        // If the node is initially syncing, ignore transactions
        if self.network.is_initially_syncing() {
            return
        }
        if self.network.tx_gossip_disabled() {
            return
        }

        // get handle to peer's session, if the session is still active
        let Some(peer) = self.peers.get_mut(&peer_id) else {
            trace!(
                peer_id = format!("{peer_id:#}"),
                ?msg,
                "discarding announcement from inactive peer"
            );

            return
        };
        let client = peer.client_version.clone();

        if msg.is_empty() {
            self.report_peer(peer_id, ReputationChangeKind::BadAnnouncement);
            return
        }

        let Ok(mut announcement) =
            TransactionAnnouncement::from_message(&msg, &mut self.announcement_hashes)
        else {
            self.report_peer(peer_id, ReputationChangeKind::BadAnnouncement);
            return
        };
        let has_duplicates = announcement.len() != msg.len();

        // keep track of the transactions the peer knows
        let count_txns_already_seen_by_peer =
            msg.iter_hashes().filter(|hash| !peer.seen_transactions.insert(**hash)).count();
        drop(msg);
        if count_txns_already_seen_by_peer > 0 {
            // this may occur if transactions are sent or announced to a peer, at the same time as
            // the peer sends/announces those hashes to us. this is because, marking
            // txns as seen by a peer is done optimistically upon sending them to the
            // peer.
            self.metrics.messages_with_hashes_already_seen_by_peer.increment(1);
            self.metrics
                .occurrences_hash_already_seen_by_peer
                .increment(count_txns_already_seen_by_peer as u64);

            trace!(target: "net::tx",
                %count_txns_already_seen_by_peer,
                peer_id=format!("{peer_id:#}"),
                ?client,
                "Peer sent hashes that have already been marked as seen by peer"
            );

            self.report_already_seen(peer_id);
        }

        if has_duplicates {
            self.report_peer(peer_id, ReputationChangeKind::BadAnnouncement);
        }

        // Exclude pending imports and known bad transactions before counting types and applying
        // policy. These cheap checks shrink the batch before acquiring the pool lock.
        let mut should_report_peer = false;
        let mut tx_types_counter = TxTypesCounter::default();
        let has_eth68_metadata = announcement.version().has_eth68_metadata();

        announcement.retain(|tx| {
            if self.transactions_by_peers.contains_key(&tx.hash) ||
                self.bad_imports.contains(&tx.hash)
            {
                return false
            }
            let (ty_byte, size_val) = tx.metadata.map_or((0, 0), |metadata| {
                match TxType::try_from(metadata.tx_type) {
                    Ok(tx_type) => tx_types_counter.increase_by_tx_type(tx_type),
                    Err(_) => tx_types_counter.increase_other(),
                }
                (metadata.tx_type, metadata.size)
            });

            let decision = self
                .policies
                .announcement_filter()
                .decide_on_announcement(ty_byte, &tx.hash, size_val);

            match decision {
                AnnouncementAcceptance::Accept => true,
                AnnouncementAcceptance::Ignore => false,
                AnnouncementAcceptance::Reject { penalize_peer } => {
                    if penalize_peer {
                        should_report_peer = true;
                    }
                    false
                }
            }
        });

        if has_eth68_metadata {
            self.announced_tx_types_metrics.update_eth68_announcement_metrics(tx_types_counter);
        }

        if should_report_peer {
            self.report_peer(peer_id, ReputationChangeKind::BadAnnouncement);
        }

        // Filter out hashes already in the pool.
        //
        // known txns have already been successfully fetched or received over gossip.
        //
        // most hashes will be filtered out here since the mempool protocol is a gossip
        // protocol, healthy peers will send many of the same hashes.
        //
        let hashes_count_pre_pool_filter = announcement.len();
        self.pool.retain_unknown(&mut announcement);
        if hashes_count_pre_pool_filter > announcement.len() {
            let already_known_hashes_count = hashes_count_pre_pool_filter - announcement.len();
            self.metrics
                .occurrences_hashes_already_in_pool
                .increment(already_known_hashes_count as u64);
        }

        if announcement.is_empty() {
            // nothing to request
            return
        }

        trace!(target: "net::tx::propagation",
            peer_id=format!("{peer_id:#}"),
            hashes_len=announcement.len(),
            msg_version=%announcement.version(),
            client_version=%client,
            "received unknown hashes in announcement from peer"
        );

        // Queue hashes for fetching; requests are sent when the manager is polled.
        self.transaction_fetcher.on_announcement(peer_id, announcement);
    }
}

impl<Pool, N> TransactionsManager<Pool, N>
where
    Pool: TransactionPool + Unpin + 'static,
    N: NetworkPrimitives<
        BroadcastedTransaction: SignedTransaction,
        PooledTransaction: SignedTransaction,
    >,
    Pool::Transaction:
        PoolTransaction<Consensus = N::BroadcastedTransaction, Pooled = N::PooledTransaction>,
{
    /// Invoked when transactions in the local mempool are considered __pending__.
    ///
    /// When a transaction in the local mempool is moved to the pending pool, we propagate them to
    /// connected peers over network using the `Transactions` and `NewPooledTransactionHashes`
    /// messages. The Transactions message relays complete transaction objects and is typically
    /// sent to a small, random fraction of connected peers.
    ///
    /// All other peers receive a notification of the transaction hash and can request the
    /// complete transaction object if it is unknown to them. The dissemination of complete
    /// transactions to a fraction of peers usually ensures that all nodes receive the transaction
    /// and won't need to request it.
    fn on_new_pending_transactions(&mut self, hashes: Vec<TxHash>) {
        // We intentionally do not gate this on initial sync.
        // During initial sync we skip importing tx announcements from peers in
        // `on_new_pooled_transaction_hashes`, so transactions reaching this path are local.
        if self.network.tx_gossip_disabled() {
            return
        }

        trace!(target: "net::tx", num_hashes=?hashes.len(), "Start propagating transactions");

        self.propagate_all(hashes);
    }

    /// Propagate the full transactions to a specific peer.
    ///
    /// Returns the propagated transactions.
    fn propagate_full_transactions_to_peer(
        &mut self,
        txs: Vec<TxHash>,
        peer_id: PeerId,
        propagation_mode: PropagationMode,
    ) -> Option<PropagatedTransactions> {
        let peer = self.peers.get_mut(&peer_id)?;
        trace!(target: "net::tx", ?peer_id, "Propagating transactions to peer");
        let mut propagated = PropagatedTransactions::default();

        // filter all transactions unknown to the peer
        let mut full_transactions = FullTransactionsBuilder::new(peer.version);

        let to_propagate = self.pool.get_all(txs).into_iter().map(PropagateTransaction::pool_tx);

        if propagation_mode.is_forced() {
            // skip cache check if forced
            full_transactions.extend(to_propagate);
        } else {
            // Iterate through the transactions to propagate and fill the hashes and full
            // transaction
            for tx in to_propagate {
                if !peer.seen_transactions.contains(tx.tx_hash()) {
                    // Only include if the peer hasn't seen the transaction
                    full_transactions.push(&tx);
                }
            }
        }

        if full_transactions.is_empty() {
            // nothing to propagate
            return None
        }

        let PropagateTransactions { pooled, full } = full_transactions.build();

        // send hashes if any
        if let Some(new_pooled_hashes) = pooled {
            for hash in new_pooled_hashes.iter_hashes().copied() {
                propagated.record(hash, PropagateKind::Hash(peer_id));
                // mark transaction as seen by peer
                peer.seen_transactions.insert(hash);
            }

            // send hashes of transactions
            self.network.send_transactions_hashes(peer_id, new_pooled_hashes);
        }

        // send full transactions, if any
        if let Some(new_full_transactions) = full {
            for hash in new_full_transactions.iter_hashes() {
                propagated.record(*hash, PropagateKind::Full(peer_id));
                // mark transaction as seen by peer
                peer.seen_transactions.insert(*hash);
            }

            // send full transactions
            self.network.send_broadcast_pool_transactions(peer_id, new_full_transactions);
        }

        // Update propagated transactions metrics
        self.metrics.propagated_transactions.increment(propagated.len() as u64);

        Some(propagated)
    }

    /// Propagate the transaction hashes to the given peer
    ///
    /// Note: This will only send the hashes for transactions that exist in the pool.
    fn propagate_hashes_to(
        &mut self,
        hashes: Vec<TxHash>,
        peer_id: PeerId,
        propagation_mode: PropagationMode,
    ) {
        trace!(target: "net::tx", "Start propagating transactions as hashes");

        // This fetches a transactions from the pool, including the blob transactions, which are
        // only ever sent as hashes.
        let propagated = {
            let Some(peer) = self.peers.get_mut(&peer_id) else {
                // no such peer
                return
            };

            let to_propagate =
                self.pool.get_all(hashes).into_iter().map(PropagateTransaction::pool_tx);

            let mut propagated = PropagatedTransactions::default();

            // check if transaction is known to peer
            let mut hashes = PooledTransactionsHashesBuilder::new(peer.version);

            if propagation_mode.is_forced() {
                hashes.extend(to_propagate)
            } else {
                for tx in to_propagate {
                    if !peer.seen_transactions.contains(tx.tx_hash()) {
                        // Include if the peer hasn't seen it
                        hashes.push(&tx);
                    }
                }
            }

            let new_pooled_hashes = hashes.build();

            if new_pooled_hashes.is_empty() {
                // nothing to propagate
                return
            }

            for hash in new_pooled_hashes.iter_hashes().copied() {
                propagated.record(hash, PropagateKind::Hash(peer_id));
                peer.seen_transactions.insert(hash);
            }

            trace!(target: "net::tx::propagation", ?peer_id, ?new_pooled_hashes, "Propagating transactions to peer");

            // send hashes of transactions
            self.network.send_transactions_hashes(peer_id, new_pooled_hashes);

            // Update propagated transactions metrics
            self.metrics.propagated_transactions.increment(propagated.len() as u64);

            propagated
        };

        // notify pool so events get fired
        self.pool.on_propagated(propagated);
    }

    /// Propagate the transactions to all connected peers either as full objects or hashes.
    ///
    /// The message for new pooled hashes depends on the negotiated version of the stream.
    /// See [`NewPooledTransactionHashes`]
    ///
    /// Note: EIP-4844 are disallowed from being broadcast in full and are only ever sent as hashes, see also <https://eips.ethereum.org/EIPS/eip-4844#networking>.
    fn propagate_transactions(
        &mut self,
        to_propagate: Vec<PropagateTransaction>,
        propagation_mode: PropagationMode,
    ) -> PropagatedTransactions {
        let mut propagated = PropagatedTransactions::default();
        if self.network.tx_gossip_disabled() {
            return propagated
        }

        // send full transactions to a set of the connected peers based on the configured mode
        let max_num_full = self.config.propagation_mode.full_peer_count(self.peers.len());

        // Note: Assuming ~random~ order due to random state of the peers map hasher
        let mut num_full_peers = 0;
        for (peer_id, peer) in &mut self.peers {
            if !self.policies.propagation_policy().can_propagate(peer) {
                // skip peers we should not propagate to
                continue
            }

            // determine whether to send full tx objects or hashes.
            let mut builder = if num_full_peers < max_num_full {
                num_full_peers += 1;
                PropagateTransactionsBuilder::full(peer.version, to_propagate.len())
            } else {
                PropagateTransactionsBuilder::pooled(peer.version, to_propagate.len())
            };

            // Transactions are optimistically marked as seen by the peer when included in the
            // message, see `PeerMetadata::seen_transactions`.
            if propagation_mode.is_forced() {
                for tx in &to_propagate {
                    peer.seen_transactions.insert(*tx.tx_hash());
                    builder.push(tx);
                }
            } else {
                // Iterate through the transactions to propagate and fill the hashes and full
                // transaction lists, before deciding whether or not to send full transactions to
                // the peer.
                for tx in &to_propagate {
                    // Only include the transaction if the peer hasn't seen it yet
                    if peer.seen_transactions.insert(*tx.tx_hash()) {
                        builder.push(tx);
                    }
                }
            }

            if builder.is_empty() {
                trace!(target: "net::tx", ?peer_id, "Nothing to propagate to peer; has seen all transactions");
                continue
            }

            let PropagateTransactions { pooled, full } = builder.build();

            // send hashes if any
            if let Some(mut new_pooled_hashes) = pooled {
                // Unhappy path: too many hashes for a single message. This should not happen
                // during regular propagation, which is capped at the soft limit per batch, and
                // is only reachable via manual propagation commands with oversized batches.
                if new_pooled_hashes.len() >
                    SOFT_LIMIT_COUNT_HASHES_IN_NEW_POOLED_TRANSACTIONS_BROADCAST_MESSAGE
                {
                    // hashes that exceed the limit are not sent, so they must not be tracked as
                    // seen by the peer
                    for hash in new_pooled_hashes
                        .iter_hashes()
                        .skip(SOFT_LIMIT_COUNT_HASHES_IN_NEW_POOLED_TRANSACTIONS_BROADCAST_MESSAGE)
                    {
                        peer.seen_transactions.remove(hash);
                    }
                    new_pooled_hashes.truncate(
                        SOFT_LIMIT_COUNT_HASHES_IN_NEW_POOLED_TRANSACTIONS_BROADCAST_MESSAGE,
                    );
                }

                for hash in new_pooled_hashes.iter_hashes().copied() {
                    propagated.record(hash, PropagateKind::Hash(*peer_id));
                }

                trace!(target: "net::tx", ?peer_id, num_txs=?new_pooled_hashes.len(), "Propagating tx hashes to peer");

                // send hashes of transactions
                self.network.send_transactions_hashes(*peer_id, new_pooled_hashes);
            }

            // send full transactions, if any
            if let Some(new_full_transactions) = full {
                for hash in new_full_transactions.iter_hashes() {
                    propagated.record(*hash, PropagateKind::Full(*peer_id));
                }

                trace!(target: "net::tx", ?peer_id, num_txs=?new_full_transactions.len(), "Propagating full transactions to peer");

                // send full transactions
                self.network.send_broadcast_pool_transactions(*peer_id, new_full_transactions);
            }
        }

        // Update propagated transactions metrics
        self.metrics.propagated_transactions.increment(propagated.len() as u64);

        propagated
    }

    /// Propagates the given transactions to the peers
    ///
    /// This fetches all transaction from the pool, including the 4844 blob transactions but
    /// __without__ their sidecar, because 4844 transactions are only ever announced as hashes.
    fn propagate_all(&mut self, hashes: Vec<TxHash>) {
        if self.peers.is_empty() {
            // nothing to propagate
            return
        }
        let propagated = self.propagate_transactions(
            self.pool.get_all(hashes).into_iter().map(PropagateTransaction::pool_tx).collect(),
            PropagationMode::Basic,
        );

        // notify pool so events get fired
        self.pool.on_propagated(propagated);
    }

    /// Request handler for an incoming request for transactions
    fn on_get_pooled_transactions(
        &mut self,
        peer_id: PeerId,
        request: GetPooledTransactions,
        response: oneshot::Sender<RequestResult<PooledTransactions<N::PooledTransaction>>>,
    ) {
        // fast exit if gossip is disabled
        if self.network.tx_gossip_disabled() {
            let _ = response.send(Ok(PooledTransactions::default()));
            return
        }
        if let Some(peer) = self.peers.get_mut(&peer_id) {
            // don't serve pending transactions to peers the policy doesn't propagate to
            if !self.policies.propagation_policy().can_propagate(peer) {
                let _ = response.send(Ok(PooledTransactions::default()));
                return
            }

            let transactions = self.pool.get_pooled_transaction_elements(
                request.0,
                GetPooledTransactionLimit::ResponseSizeSoftLimit(
                    self.config
                        .transaction_fetcher_config
                        .soft_limit_byte_size_pooled_transactions_response,
                ),
            );
            trace!(target: "net::tx::propagation", sent_txs=?transactions.iter().map(|tx| tx.tx_hash()), "Sending requested transactions to peer");

            // we sent a response at which point we assume that the peer is aware of the
            // transactions
            peer.seen_transactions.extend(transactions.iter().map(|tx| *tx.tx_hash()));

            let resp = PooledTransactions(transactions);
            let _ = response.send(Ok(resp));
        }
    }

    /// Handles a command received from a detached [`TransactionsHandle`]
    fn on_command(&mut self, cmd: TransactionsCommand<N>) {
        match cmd {
            TransactionsCommand::PropagateHash(hash) => {
                self.on_new_pending_transactions(vec![hash])
            }
            TransactionsCommand::PropagateHashesTo(hashes, peer) => {
                self.propagate_hashes_to(hashes, peer, PropagationMode::Forced)
            }
            TransactionsCommand::GetActivePeers(tx) => {
                let peers = self.peers.keys().copied().collect::<HashSet<_>>();
                tx.send(peers).ok();
            }
            TransactionsCommand::PropagateTransactionsTo(txs, peer) => {
                if let Some(propagated) =
                    self.propagate_full_transactions_to_peer(txs, peer, PropagationMode::Forced)
                {
                    self.pool.on_propagated(propagated);
                }
            }
            TransactionsCommand::PropagateTransactions(txs) => self.propagate_all(txs),
            TransactionsCommand::BroadcastTransactions(txs) => {
                let propagated = self.propagate_transactions(txs, PropagationMode::Forced);
                self.pool.on_propagated(propagated);
            }
            TransactionsCommand::GetTransactionHashes { peers, tx } => {
                let mut res = HashMap::with_capacity_and_hasher(peers.len(), Default::default());
                for peer_id in peers {
                    let hashes = self
                        .peers
                        .get(&peer_id)
                        .map(|peer| peer.seen_transactions.iter().copied().collect::<B256Set>())
                        .unwrap_or_default();
                    res.insert(peer_id, hashes);
                }
                tx.send(res).ok();
            }
            TransactionsCommand::GetPeerSender { peer_id, peer_request_sender } => {
                let sender = self.peers.get(&peer_id).map(|peer| peer.request_tx.clone());
                peer_request_sender.send(sender).ok();
            }
        }
    }

    /// Handles session establishment and peer transactions initialization.
    ///
    /// This is invoked when a new session is established.
    fn handle_peer_session(
        &mut self,
        info: SessionInfo,
        messages: PeerRequestSender<PeerRequest<N>>,
    ) {
        let SessionInfo { peer_id, client_version, version, .. } = info;

        // Insert a new peer into the peerset.
        let peer = PeerMetadata::<N>::new(
            messages,
            version,
            client_version,
            self.config.max_transactions_seen_by_peer_history,
            info.peer_kind,
        );
        let peer = match self.peers.entry(peer_id) {
            Entry::Occupied(mut entry) => {
                entry.insert(peer);
                entry.into_mut()
            }
            Entry::Vacant(entry) => entry.insert(peer),
        };

        self.policies.propagation_policy_mut().on_session_established(peer);

        // Send a `NewPooledTransactionHashes` to the peer with up to
        // `SOFT_LIMIT_COUNT_HASHES_IN_NEW_POOLED_TRANSACTIONS_BROADCAST_MESSAGE`
        // transactions in the pool.
        if self.network.is_initially_syncing() || self.network.tx_gossip_disabled() {
            trace!(target: "net::tx", ?peer_id, "Skipping transaction broadcast: node syncing or gossip disabled");
            return
        }

        // skip peers we should not propagate to
        if !self.policies.propagation_policy().can_propagate(peer) {
            return
        }

        // Get transactions to broadcast
        let pooled_txs = self.pool.pooled_transactions_max(
            SOFT_LIMIT_COUNT_HASHES_IN_NEW_POOLED_TRANSACTIONS_BROADCAST_MESSAGE,
        );
        if pooled_txs.is_empty() {
            trace!(target: "net::tx", ?peer_id, "No transactions in the pool to broadcast");
            return;
        }

        // Build and send transaction hashes message
        let mut msg_builder = PooledTransactionsHashesBuilder::new(version);
        for pooled_tx in pooled_txs {
            peer.seen_transactions.insert(*pooled_tx.hash());
            msg_builder.push_pooled(pooled_tx);
        }

        debug!(target: "net::tx", ?peer_id, tx_count = msg_builder.len(), "Broadcasting transaction hashes");
        let msg = msg_builder.build();
        self.network.send_transactions_hashes(peer_id, msg);
    }

    /// Handles a received event related to common network events.
    fn on_network_event(&mut self, event_result: NetworkEvent<PeerRequest<N>>) {
        match event_result {
            NetworkEvent::Peer(PeerEvent::SessionClosed { peer_id, .. }) => {
                self.on_peer_session_closed(&peer_id);
            }
            NetworkEvent::ActivePeerSession { info, messages } => {
                // process active peer session and broadcast available transaction from the pool
                self.handle_peer_session(info, messages);
            }
            NetworkEvent::Peer(PeerEvent::SessionEstablished(info)) => {
                let peer_id = info.peer_id;
                // get messages from existing peer
                let messages = match self.peers.get(&peer_id) {
                    Some(p) => p.request_tx.clone(),
                    None => {
                        debug!(target: "net::tx", ?peer_id, "No peer request sender found");
                        return;
                    }
                };
                self.handle_peer_session(info, messages);
            }
            _ => {}
        }
    }

    /// Returns true if the ingress policy allows processing messages from the given peer.
    fn accepts_incoming_from(&self, peer_id: &PeerId) -> bool {
        if self.config.ingress_policy.allows_all() {
            return true;
        }
        let Some(peer) = self.peers.get(peer_id) else {
            return false;
        };
        self.config.ingress_policy.allows(peer.peer_kind())
    }

    /// Handles dedicated transaction events related to the `eth` protocol.
    fn on_network_tx_event(&mut self, event: NetworkTransactionEvent<N>) {
        match event {
            NetworkTransactionEvent::IncomingTransactions { peer_id, msg } => {
                if !self.accepts_incoming_from(&peer_id) {
                    trace!(target: "net::tx", peer_id=format!("{peer_id:#}"), policy=?self.config.ingress_policy, "Ignoring full transactions from peer blocked by ingress policy");
                    return;
                }

                // ensure we didn't receive any blob transactions as these are disallowed to be
                // broadcasted in full

                let has_blob_txs = msg.has_eip4844();

                let non_blob_txs = msg
                    .into_iter()
                    .map(N::PooledTransaction::try_from)
                    .filter_map(Result::ok)
                    .collect();

                self.import_transactions(peer_id, non_blob_txs, TransactionSource::Broadcast);

                if has_blob_txs {
                    debug!(target: "net::tx", ?peer_id, "received bad full blob transaction broadcast");
                    self.report_peer_bad_transactions(peer_id);
                }
            }
            NetworkTransactionEvent::IncomingPooledTransactionHashes { peer_id, msg } => {
                if !self.accepts_incoming_from(&peer_id) {
                    trace!(target: "net::tx", peer_id=format!("{peer_id:#}"), policy=?self.config.ingress_policy, "Ignoring transaction hashes from peer blocked by ingress policy");
                    return;
                }
                self.on_new_pooled_transaction_hashes(peer_id, msg)
            }
            NetworkTransactionEvent::GetPooledTransactions { peer_id, request, response } => {
                self.on_get_pooled_transactions(peer_id, request, response)
            }
            NetworkTransactionEvent::GetTransactionsHandle(response) => {
                let _ = response.send(Some(self.handle()));
            }
        }
    }

    /// Starts the import process for the given transactions.
    ///
    /// Callers must check import capacity before consuming a fetch response. Responses are
    /// admitted in full and may overshoot the soft limit by one bounded response; broadcasts
    /// are truncated to their available capacity.
    fn import_transactions(
        &mut self,
        peer_id: PeerId,
        transactions: PooledTransactions<N::PooledTransaction>,
        source: TransactionSource,
    ) {
        // If the node is pipeline syncing, ignore transactions
        if self.network.is_initially_syncing() {
            return
        }
        if self.network.tx_gossip_disabled() {
            return
        }

        let (version, client_version) = match &source {
            TransactionSource::Broadcast => {
                let Some(peer) = self.peers.get(&peer_id) else { return };
                (peer.version(), peer.client_version.clone())
            }
            TransactionSource::Response { version, client_version } => {
                (*version, client_version.clone())
            }
        };
        let is_broadcast = source.is_broadcast();
        let mut transactions = transactions.0;

        // Fetched batches may exceed the soft limit: their request size is bounded, and the
        // caller stops consuming responses until capacity becomes available again.
        if is_broadcast {
            let (capacity, byte_capacity) = self.remaining_broadcast_import_capacity();
            let mut bytes = 0usize;
            let admitted = transactions
                .iter()
                .take(capacity)
                .take_while(|tx| {
                    bytes = bytes.saturating_add(tx.encode_2718_len());
                    bytes <= byte_capacity
                })
                .count();
            if transactions.len() > admitted {
                self.metrics
                    .skipped_transactions_pending_pool_imports_at_capacity
                    .increment((transactions.len() - admitted) as u64);
                transactions.truncate(admitted);
            }
        }

        if transactions.is_empty() {
            return
        }

        let start = Instant::now();

        // Stop fetching transactions received through either broadcasts or responses.
        self.transaction_fetcher
            .on_transactions_received(transactions.iter().map(|tx| tx.tx_hash()));

        // track that the peer knows these transaction, but only if this is a new broadcast.
        // If we received the transactions as the response to our `GetPooledTransactions``
        // requests (based on received `NewPooledTransactionHashes`) then we already
        // recorded the hashes as seen by this peer in `Self::on_new_pooled_transaction_hashes`.
        let mut num_already_seen_by_peer = 0;
        if is_broadcast && let Some(peer) = self.peers.get_mut(&peer_id) {
            for tx in &transactions {
                if !peer.seen_transactions.insert(*tx.tx_hash()) {
                    num_already_seen_by_peer += 1;
                }
            }
        }

        // Eth/72 `PooledTransactions` responses elide blob payloads from type 3 transactions per
        // EIP-8070, so their sidecars can never validate. Drop them before touching the pool;
        // geth equivalently diverts these bodies into a buffer that is completed with cells
        // fetched via `GetCells`, which is not implemented yet.
        if version.is_eth72() {
            let len_before = transactions.len();
            transactions.retain(|tx| !tx.is_eip4844());
            let dropped = len_before - transactions.len();
            if dropped > 0 {
                trace!(target: "net::tx",
                    peer_id=format!("{peer_id:#}"),
                    dropped,
                    "dropped blob transactions from eth72 response, cell fetching not implemented"
                );
            }
        }

        // tracks the quality of the given transactions
        let mut has_bad_transactions = false;

        // 1. Remove known, already-tracked, and invalid transactions first since these are
        // cheap in-memory checks against local maps
        transactions.retain(|tx| {
            if let Entry::Occupied(mut entry) = self.transactions_by_peers.entry(*tx.tx_hash()) {
                let peers = entry.get_mut();
                if !peers.contains(&peer_id) {
                    peers.push(peer_id);
                }
                return false
            }
            if self.bad_imports.contains(tx.tx_hash()) {
                trace!(target: "net::tx",
                    peer_id=format!("{peer_id:#}"),
                    hash=%tx.tx_hash(),
                    %client_version,
                    "received a known bad transaction from peer"
                );
                has_bad_transactions = true;
                return false;
            }
            true
        });

        // 2. filter out txns already inserted into pool
        let txns_count_pre_pool_filter = transactions.len();
        self.pool.retain_unknown(&mut transactions);
        if txns_count_pre_pool_filter > transactions.len() {
            let already_known_txns_count = txns_count_pre_pool_filter - transactions.len();
            self.metrics
                .occurrences_transactions_already_in_pool
                .increment(already_known_txns_count as u64);
        }

        let txs_len = transactions.len();

        let recover = |tx| {
            let recovered = Pool::Transaction::try_recover_with_cache_opt(
                tx,
                self.sender_recovery_cache.as_ref(),
            );
            match recovered {
                Ok(tx) => Some(tx),
                Err(badtx) => {
                    trace!(target: "net::tx",
                        peer_id=format!("{peer_id:#}"),
                        hash=%badtx.tx_hash(),
                        client_version=%client_version,
                        "failed ecrecovery for transaction"
                    );
                    None
                }
            }
        };

        let new_txs = transactions.into_par_iter().filter_map(recover).collect::<Vec<_>>();

        has_bad_transactions |= new_txs.len() != txs_len;

        // Record the transactions as seen by the peer
        for tx in &new_txs {
            self.transactions_by_peers.insert(*tx.hash(), smallvec::smallvec![peer_id]);
        }

        // 3. import new transactions as a batch to minimize lock contention on the underlying
        // pool
        if !new_txs.is_empty() {
            let pool = self.pool.clone();
            // update metrics
            let metric_pending_pool_imports = self.metrics.pending_pool_imports.clone();
            metric_pending_pool_imports.increment(new_txs.len() as f64);
            // includes blob sidecars, which stay in memory until the import completes
            let bytes = new_txs.iter().map(PoolTransaction::encoded_length).sum::<usize>();
            let metric_pending_pool_imports_bytes = self.metrics.pending_pool_imports_bytes.clone();
            metric_pending_pool_imports_bytes.increment(bytes as f64);

            // update self-monitoring info
            self.pending_pool_imports_info
                .pending_pool_imports
                .fetch_add(new_txs.len(), Ordering::Relaxed);
            let tx_manager_info_pending_pool_imports =
                self.pending_pool_imports_info.pending_pool_imports.clone();
            self.pending_pool_imports_info
                .pending_pool_imports_bytes
                .fetch_add(bytes, Ordering::Relaxed);
            let tx_manager_info_pending_pool_imports_bytes =
                self.pending_pool_imports_info.pending_pool_imports_bytes.clone();

            trace!(target: "net::tx::propagation", new_txs_len=?new_txs.len(), "Importing new transactions");
            let import = Box::pin(async move {
                let added = new_txs.len();
                let res = pool.add_external_transactions(new_txs).await;

                // update metrics
                metric_pending_pool_imports.decrement(added as f64);
                metric_pending_pool_imports_bytes.decrement(bytes as f64);
                // update self-monitoring info
                tx_manager_info_pending_pool_imports.fetch_sub(added, Ordering::Relaxed);
                tx_manager_info_pending_pool_imports_bytes.fetch_sub(bytes, Ordering::Relaxed);

                res
            });

            self.pool_imports.push(import);
        }

        if num_already_seen_by_peer > 0 {
            self.metrics.messages_with_transactions_already_seen_by_peer.increment(1);
            self.metrics
                .occurrences_of_transaction_already_seen_by_peer
                .increment(num_already_seen_by_peer);
            trace!(target: "net::tx", num_txs=%num_already_seen_by_peer, ?peer_id, client=%client_version, "Peer sent already seen transactions");
        }

        if has_bad_transactions {
            // peer sent us invalid transactions
            self.report_peer_bad_transactions(peer_id)
        }

        if num_already_seen_by_peer > 0 {
            self.report_already_seen(peer_id);
        }

        self.metrics.pool_import_prepare_duration.record(start.elapsed());
    }

    /// Processes a [`FetchEvent`].
    fn on_fetch_event(&mut self, fetch_event: FetchEvent<N::PooledTransaction>) {
        match fetch_event {
            FetchEvent::TransactionsFetched {
                peer_id,
                transactions,
                report_peer,
                version,
                client_version,
            } => {
                self.import_transactions(
                    peer_id,
                    transactions,
                    TransactionSource::Response { version, client_version },
                );
                if report_peer {
                    self.report_peer(peer_id, ReputationChangeKind::BadTransactions);
                }
            }
            FetchEvent::FetchError { peer_id, error } => {
                trace!(target: "net::tx", ?peer_id, %error, "requesting transactions from peer failed");
                self.on_request_error(peer_id, error);
            }
            FetchEvent::EmptyResponse { peer_id } => {
                trace!(target: "net::tx", ?peer_id, "peer returned empty response");
            }
        }
    }
}

/// An endless future. Preemption ensure that future is non-blocking, nonetheless. See
/// [`crate::NetworkManager`] for more context on the design pattern.
///
/// This should be spawned or used as part of `tokio::select!`.
//
// spawned in `NodeConfig::start_network`(reth_node_core::NodeConfig) and
// `NetworkConfig::start_network`(reth_network::NetworkConfig)
impl<
        Pool: TransactionPool + Unpin + 'static,
        N: NetworkPrimitives<
            BroadcastedTransaction: SignedTransaction,
            PooledTransaction: SignedTransaction,
        >,
    > Future for TransactionsManager<Pool, N>
where
    Pool::Transaction:
        PoolTransaction<Consensus = N::BroadcastedTransaction, Pooled = N::PooledTransaction>,
{
    type Output = ();

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let start = Instant::now();
        let mut poll_durations = TxManagerPollDurations::default();

        let this = self.get_mut();

        // All streams are polled until their corresponding budget is exhausted, then we manually
        // yield back control to tokio. See `NetworkManager` for more context on the design
        // pattern.

        // Advance network/peer related events (update peers map).
        let maybe_more_network_events = metered_poll_nested_stream_with_budget!(
            poll_durations.acc_network_events,
            "net::tx",
            "Network events stream",
            DEFAULT_BUDGET_TRY_DRAIN_STREAM,
            this.network_events.poll_next_unpin(cx),
            |event| this.on_network_event(event)
        );

        // Advance incoming transaction events (stream new txns/announcements from
        // network manager and queue for import to pool/fetch txns).
        //
        // Announcements are queued in the transaction fetcher, requests for them are sent
        // further below.
        //
        // Decoded broadcasts may exceed the available import capacity. Admission truncates
        // them before filtering and sender recovery, reserving capacity for fetched transactions.
        let maybe_more_tx_events = metered_poll_nested_stream_with_budget!(
            poll_durations.acc_tx_events,
            "net::tx",
            "Network transaction events stream",
            DEFAULT_BUDGET_TRY_DRAIN_NETWORK_TRANSACTION_EVENTS,
            this.transaction_events.poll_next_unpin(cx),
            |event: NetworkTransactionEvent<N>| this.on_network_tx_event(event),
        );

        // Advance inflight fetch requests (flush transaction fetcher and queue for
        // import to pool).
        //
        // Pace response consumption with the soft import limit. Dispatch checks capacity but
        // does not reserve it for outstanding requests, so concurrent responses can exceed it.
        // Admit one complete response while capacity remains (at most 256 transactions), then
        // pause here if it fills or overshoots the limit. Existing requests continue in flight;
        // completed responses wait in their channels until pool imports free capacity.
        let mut maybe_more_tx_fetch_events = metered_poll_nested_stream_with_budget!(
            poll_durations.acc_fetch_events,
            "net::tx",
            "Transaction fetch events stream",
            DEFAULT_BUDGET_TRY_DRAIN_STREAM,
            if this.has_capacity_for_pending_pool_imports() {
                this.transaction_fetcher.poll_next_unpin(cx)
            } else {
                Poll::Pending
            },
            |event| this.on_fetch_event(event),
        );

        // Remember whether response polling stopped at capacity so imports completing below
        // can trigger another poll even if the waiting response has not registered a waker yet.
        let fetcher_paused_at_capacity = !this.has_capacity_for_pending_pool_imports();

        // Advance admitted batches and free capacity for more responses.
        let maybe_more_pool_imports = metered_poll_nested_stream_with_budget!(
            poll_durations.acc_pending_imports,
            "net::tx",
            "Batched pool imports stream",
            DEFAULT_BUDGET_TRY_DRAIN_PENDING_POOL_IMPORTS,
            this.pool_imports.poll_next_unpin(cx),
            |batch_results| this.on_batch_import_result(batch_results)
        );

        // Advances new __pending__ transactions, transactions that were successfully inserted into
        // pending set in pool (are valid), and propagates them (inform peers which
        // transactions we have seen).
        //
        // This is polled after pool imports so transactions that became pending in this poll
        // iteration are propagated immediately, instead of waiting for the task to be woken
        // again.
        //
        // We try to drain this to batch the transactions in a single message.
        //
        // We don't expect this buffer to be large, since only pending transactions are
        // emitted here.
        let mut new_txs = Vec::new();
        let maybe_more_pending_txns = match this.pending_transactions.poll_recv_many(
            cx,
            &mut new_txs,
            SOFT_LIMIT_COUNT_HASHES_IN_NEW_POOLED_TRANSACTIONS_BROADCAST_MESSAGE,
        ) {
            Poll::Ready(count) => {
                if count == SOFT_LIMIT_COUNT_HASHES_IN_NEW_POOLED_TRANSACTIONS_BROADCAST_MESSAGE {
                    // we filled the entire buffer capacity and need to try again on the next poll
                    // immediately
                    true
                } else {
                    // try once more, because mostlikely the channel is now empty and the waker is
                    // registered if this is pending, if we filled additional hashes, we poll again
                    // on the next iteration
                    let limit =
                        SOFT_LIMIT_COUNT_HASHES_IN_NEW_POOLED_TRANSACTIONS_BROADCAST_MESSAGE -
                            new_txs.len();
                    this.pending_transactions.poll_recv_many(cx, &mut new_txs, limit).is_ready()
                }
            }
            Poll::Pending => false,
        };
        if !new_txs.is_empty() {
            this.on_new_pending_transactions(new_txs);
        }

        // Dispatch only below the soft import limit. Each request is capped independently;
        // outstanding requests do not reserve capacity against other peers.
        duration_metered_exec!(
            {
                // Peers whose session channel was full are retried on the next poll, which any
                // other event triggers.
                let budget = this.remaining_pool_import_capacity();
                if budget > 0 && this.transaction_fetcher.dispatch(&this.peers, budget) > 0 {
                    // poll the fetcher again so the new inflight requests register their wakers
                    maybe_more_tx_fetch_events = true;
                }
            },
            poll_durations.acc_pending_fetch
        );

        // Advance commands (propagate/fetch/serve txns).
        let maybe_more_commands = metered_poll_nested_stream_with_budget!(
            poll_durations.acc_cmds,
            "net::tx",
            "Commands channel",
            DEFAULT_BUDGET_TRY_DRAIN_STREAM,
            this.command_rx.poll_next_unpin(cx),
            |cmd| this.on_command(cmd)
        );

        this.transaction_fetcher.update_metrics();

        // Revisit the fetcher if imports freed capacity after response polling was paused.
        let resume_fetcher =
            fetcher_paused_at_capacity && this.has_capacity_for_pending_pool_imports();

        // all channels are fully drained and import futures pending
        if maybe_more_network_events ||
            maybe_more_commands ||
            maybe_more_tx_events ||
            maybe_more_tx_fetch_events ||
            maybe_more_pool_imports ||
            maybe_more_pending_txns ||
            resume_fetcher
        {
            // make sure we're woken up again
            cx.waker().wake_by_ref();
            return Poll::Pending
        }

        this.update_poll_metrics(start, poll_durations);

        Poll::Pending
    }
}

/// Represents the different modes of transaction propagation.
///
/// This enum is used to determine how transactions are propagated to peers in the network.
#[derive(Debug, Copy, Clone, Eq, PartialEq)]
enum PropagationMode {
    /// Default propagation mode.
    ///
    /// Transactions are only sent to peers that haven't seen them yet.
    Basic,
    /// Forced propagation mode.
    ///
    /// Transactions are sent to all peers regardless of whether they have been sent or received
    /// before.
    Forced,
}

impl PropagationMode {
    /// Returns `true` if the propagation kind is `Forced`.
    const fn is_forced(self) -> bool {
        matches!(self, Self::Forced)
    }
}

/// A transaction that's about to be propagated to multiple peers.
#[derive(Debug, Clone)]
struct PropagateTransaction {
    is_broadcastable_in_full: bool,
    /// Size advertised in `NewPooledTransactionHashes` metadata and used for full broadcast
    /// soft-limit accounting.
    ///
    /// This is the network encoded transaction size. For pool-backed blob transactions, this is
    /// the pool's cached encoded length, which includes the sidecar returned by
    /// `PooledTransactions`.
    propagation_size: usize,
    transaction: LazyEncodedTransaction,
}

impl PropagateTransaction {
    /// Create a new instance from a transaction supplied directly for propagation.
    ///
    /// Direct transactions use their EIP-2718 encoded length so eth/68+ hash announcements carry
    /// the same size metadata as [`NewPooledTransactionHashes68::push`] and
    /// [`NewPooledTransactionHashes72::push`].
    fn new<T: SignedTransaction>(transaction: T) -> Self {
        let is_broadcastable_in_full = transaction.is_broadcastable_in_full();
        let propagation_size = transaction.encode_2718_len();

        Self {
            is_broadcastable_in_full,
            propagation_size,
            transaction: LazyEncoded::new(transaction),
        }
    }

    /// Create a new instance from a pooled transaction.
    ///
    /// Pool transactions already cache the network encoded size used by txpool admission and
    /// pooled hash announcements. For blob transactions, this includes the sidecar size expected in
    /// a `PooledTransactions` response.
    fn pool_tx<P: PoolTransaction>(tx: Arc<ValidPoolTransaction<P>>) -> Self {
        let is_broadcastable_in_full = tx.transaction.consensus_ref().is_broadcastable_in_full();
        let propagation_size = tx.encoded_length();
        Self {
            is_broadcastable_in_full,
            propagation_size,
            transaction: LazyEncoded::new(PropagatePooledTransactionEncoder::new(tx)),
        }
    }

    fn tx_hash(&self) -> &TxHash {
        self.transaction.tx_hash()
    }

    /// Returns the network encoded size used for propagation limits and hash metadata.
    const fn propagation_size(&self) -> usize {
        self.propagation_size
    }

    fn tx_type(&self) -> u8 {
        self.transaction.ty()
    }

    const fn is_broadcastable_in_full(&self) -> bool {
        self.is_broadcastable_in_full
    }

    fn shared(&self) -> LazyEncodedTransaction {
        self.transaction.clone()
    }
}

/// A pooled transaction encoder that avoids cloning into the consensus transaction for propagation.
#[derive(Debug)]
struct PropagatePooledTransactionEncoder<P: PoolTransaction> {
    transaction: Arc<ValidPoolTransaction<P>>,
}

impl<P: PoolTransaction> PropagatePooledTransactionEncoder<P> {
    const fn new(transaction: Arc<ValidPoolTransaction<P>>) -> Self {
        Self { transaction }
    }

    fn encode_uncached(&self, out: &mut dyn BufMut) {
        (*self.transaction.transaction.consensus_ref().inner()).encode(out);
    }
}

impl<P: PoolTransaction> Encodable for PropagatePooledTransactionEncoder<P> {
    fn encode(&self, out: &mut dyn BufMut) {
        self.encode_uncached(out);
    }

    fn length(&self) -> usize {
        (*self.transaction.transaction.consensus_ref().inner()).length()
    }
}

impl<P: PoolTransaction> TxHashRef for PropagatePooledTransactionEncoder<P> {
    fn tx_hash(&self) -> &TxHash {
        self.transaction.hash()
    }
}

impl<P: PoolTransaction> Typed2718 for PropagatePooledTransactionEncoder<P> {
    fn ty(&self) -> u8 {
        self.transaction.transaction.ty()
    }
}

/// Helper type to construct the appropriate message to send to the peer based on whether the peer
/// should receive them in full or as pooled
#[derive(Debug, Clone)]
enum PropagateTransactionsBuilder {
    Pooled(PooledTransactionsHashesBuilder),
    Full(FullTransactionsBuilder),
}

impl PropagateTransactionsBuilder {
    /// Create a builder for pooled transactions with capacity for the expected number of
    /// transactions.
    fn pooled(version: EthVersion, capacity: usize) -> Self {
        Self::Pooled(PooledTransactionsHashesBuilder::with_capacity(version, capacity))
    }

    /// Create a builder that sends transactions in full and records transactions that don't fit,
    /// with capacity for the expected number of transactions.
    fn full(version: EthVersion, capacity: usize) -> Self {
        Self::Full(FullTransactionsBuilder::with_capacity(version, capacity))
    }

    /// Returns true if no transactions are recorded.
    fn is_empty(&self) -> bool {
        match self {
            Self::Pooled(builder) => builder.is_empty(),
            Self::Full(builder) => builder.is_empty(),
        }
    }

    /// Consumes the type and returns the built messages that should be sent to the peer.
    fn build(self) -> PropagateTransactions {
        match self {
            Self::Pooled(pooled) => {
                PropagateTransactions { pooled: Some(pooled.build()), full: None }
            }
            Self::Full(full) => full.build(),
        }
    }
}

impl PropagateTransactionsBuilder {
    /// Appends a transaction to the list.
    fn push(&mut self, transaction: &PropagateTransaction) {
        match self {
            Self::Pooled(builder) => builder.push(transaction),
            Self::Full(builder) => builder.push(transaction),
        }
    }
}

/// Represents how the transactions should be sent to a peer if any.
struct PropagateTransactions {
    /// The pooled transaction hashes to send.
    pooled: Option<NewPooledTransactionHashes>,
    /// The transactions to send in full.
    full: Option<BroadcastPoolTransactions>,
}

/// Helper type for constructing the full transaction message that enforces the
/// [`DEFAULT_SOFT_LIMIT_BYTE_SIZE_TRANSACTIONS_BROADCAST_MESSAGE`] for full transaction broadcast
/// and enforces other propagation rules for EIP-4844 and tracks those transactions that can't be
/// broadcasted in full.
#[derive(Debug, Clone)]
struct FullTransactionsBuilder {
    /// The soft limit to enforce for a single broadcast message of full transactions.
    total_size: usize,
    /// All transactions to be broadcasted.
    transactions: Vec<LazyEncodedTransaction>,
    /// Transactions that didn't fit into the broadcast message
    pooled: PooledTransactionsHashesBuilder,
}

impl FullTransactionsBuilder {
    /// Create a builder for the negotiated version of the peer's session
    fn new(version: EthVersion) -> Self {
        Self {
            total_size: 0,
            pooled: PooledTransactionsHashesBuilder::new(version),
            transactions: vec![],
        }
    }

    /// Create a builder with capacity for the expected number of full transactions.
    ///
    /// The overflow hashes builder remains lazily allocated since most transactions are expected
    /// to be broadcast in full.
    fn with_capacity(version: EthVersion, capacity: usize) -> Self {
        Self {
            total_size: 0,
            pooled: PooledTransactionsHashesBuilder::new(version),
            transactions: Vec::with_capacity(capacity),
        }
    }

    /// Returns whether or not any transactions are in the [`FullTransactionsBuilder`].
    fn is_empty(&self) -> bool {
        self.transactions.is_empty() && self.pooled.is_empty()
    }

    /// Returns the messages that should be propagated to the peer.
    fn build(self) -> PropagateTransactions {
        let pooled = Some(self.pooled.build()).filter(|pooled| !pooled.is_empty());
        let full =
            (!self.transactions.is_empty()).then_some(BroadcastPoolTransactions(self.transactions));
        PropagateTransactions { pooled, full }
    }

    /// Appends all transactions.
    fn extend(&mut self, txs: impl IntoIterator<Item = PropagateTransaction>) {
        for tx in txs {
            self.push(&tx)
        }
    }

    /// Append a transaction to the list of full transaction if the total message bytes size doesn't
    /// exceed the soft maximum target byte size. The limit is soft, meaning if one single
    /// transaction goes over the limit, it will be broadcasted in its own [`Transactions`]
    /// message. The same pattern is followed in filling a [`GetPooledTransactions`] request in
    /// [`TransactionFetcher::dispatch`].
    ///
    /// If the transaction is unsuitable for broadcast or would exceed the softlimit, it is appended
    /// to list of pooled transactions, (e.g. 4844 transactions).
    /// See also [`SignedTransaction::is_broadcastable_in_full`].
    fn push(&mut self, transaction: &PropagateTransaction) {
        // Do not send full 4844 transaction hashes to peers.
        //
        //  Nodes MUST NOT automatically broadcast blob transactions to their peers.
        //  Instead, those transactions are only announced using
        //  `NewPooledTransactionHashes` messages, and can then be manually requested
        //  via `GetPooledTransactions`.
        //
        // From: <https://eips.ethereum.org/EIPS/eip-4844#networking>
        if !transaction.is_broadcastable_in_full() {
            self.pooled.push(transaction);
            return
        }

        let new_size = self.total_size + transaction.propagation_size();
        if new_size > DEFAULT_SOFT_LIMIT_BYTE_SIZE_TRANSACTIONS_BROADCAST_MESSAGE &&
            self.total_size > 0
        {
            // transaction does not fit into the message
            self.pooled.push(transaction);
            return
        }

        self.total_size = new_size;
        self.transactions.push(transaction.shared());
    }
}

/// A helper type to create the pooled transactions message based on the negotiated version of the
/// session with the peer
#[derive(Debug, Clone)]
enum PooledTransactionsHashesBuilder {
    Eth66(NewPooledTransactionHashes66),
    Eth68(NewPooledTransactionHashes68),
    Eth72(NewPooledTransactionHashes72),
}

// === impl PooledTransactionsHashesBuilder ===

impl PooledTransactionsHashesBuilder {
    /// Push a transaction from the pool to the list.
    fn push_pooled<T: PoolTransaction>(&mut self, pooled_tx: Arc<ValidPoolTransaction<T>>) {
        match self {
            Self::Eth66(msg) => msg.push(*pooled_tx.hash()),
            Self::Eth68(msg) => {
                msg.hashes.push(*pooled_tx.hash());
                msg.sizes.push(pooled_tx.encoded_length());
                msg.types.push(pooled_tx.transaction.ty());
            }
            Self::Eth72(msg) => {
                msg.hashes.push(*pooled_tx.hash());
                msg.sizes.push(pooled_tx.encoded_length());
                let ty = pooled_tx.transaction.ty();
                msg.types.push(ty);
                if ty == EIP4844_TX_TYPE_ID {
                    // The pool holds the full sidecar, so every cell can be served.
                    msg.cell_mask = Some(NewPooledTransactionHashes72::ALL_CELLS_MASK);
                }
            }
        }
    }

    /// Returns whether or not any transactions are in the [`PooledTransactionsHashesBuilder`].
    fn is_empty(&self) -> bool {
        match self {
            Self::Eth66(hashes) => hashes.is_empty(),
            Self::Eth68(hashes) => hashes.is_empty(),
            Self::Eth72(hashes) => hashes.is_empty(),
        }
    }

    /// Returns the number of transactions in the builder.
    fn len(&self) -> usize {
        match self {
            Self::Eth66(hashes) => hashes.len(),
            Self::Eth68(hashes) => hashes.len(),
            Self::Eth72(hashes) => hashes.len(),
        }
    }

    /// Appends all hashes
    fn extend(&mut self, txs: impl IntoIterator<Item = PropagateTransaction>) {
        for tx in txs {
            self.push(&tx);
        }
    }

    fn push(&mut self, tx: &PropagateTransaction) {
        match self {
            Self::Eth66(msg) => msg.push(*tx.tx_hash()),
            Self::Eth68(msg) => {
                msg.hashes.push(*tx.tx_hash());
                msg.sizes.push(tx.propagation_size());
                msg.types.push(tx.tx_type());
            }
            Self::Eth72(msg) => {
                msg.hashes.push(*tx.tx_hash());
                msg.sizes.push(tx.propagation_size());
                let ty = tx.tx_type();
                msg.types.push(ty);
                if ty == EIP4844_TX_TYPE_ID {
                    // The pool holds the full sidecar, so every cell can be served.
                    msg.cell_mask = Some(NewPooledTransactionHashes72::ALL_CELLS_MASK);
                }
            }
        }
    }

    /// Create a builder for the negotiated version of the peer's session
    fn new(version: EthVersion) -> Self {
        match version {
            EthVersion::Eth66 | EthVersion::Eth67 => Self::Eth66(Default::default()),
            EthVersion::Eth68 | EthVersion::Eth69 | EthVersion::Eth70 | EthVersion::Eth71 => {
                Self::Eth68(Default::default())
            }
            EthVersion::Eth72 => Self::Eth72(Default::default()),
        }
    }

    /// Create a builder for the negotiated version of the peer's session with capacity for the
    /// expected number of hashes.
    fn with_capacity(version: EthVersion, capacity: usize) -> Self {
        match version {
            EthVersion::Eth66 | EthVersion::Eth67 => {
                Self::Eth66(NewPooledTransactionHashes66::with_capacity(capacity))
            }
            EthVersion::Eth68 | EthVersion::Eth69 | EthVersion::Eth70 | EthVersion::Eth71 => {
                Self::Eth68(NewPooledTransactionHashes68::with_capacity(capacity))
            }
            EthVersion::Eth72 => Self::Eth72(NewPooledTransactionHashes72::with_capacity(capacity)),
        }
    }

    fn build(self) -> NewPooledTransactionHashes {
        match self {
            Self::Eth66(mut msg) => {
                msg.shrink_to_fit();
                msg.into()
            }
            Self::Eth68(mut msg) => {
                msg.shrink_to_fit();
                msg.into()
            }
            Self::Eth72(mut msg) => {
                msg.shrink_to_fit();
                msg.into()
            }
        }
    }
}

/// How we received the transactions.
#[derive(Debug)]
enum TransactionSource {
    /// Transactions were broadcast to us via [`Transactions`] message.
    Broadcast,
    /// Transactions were sent as the response to a `GetPooledTransactions` request issued by us.
    Response {
        /// Session metadata remains available after the responding peer disconnects.
        version: EthVersion,
        client_version: Arc<str>,
    },
}

// === impl TransactionSource ===

impl TransactionSource {
    /// Whether the transaction were sent as broadcast.
    const fn is_broadcast(&self) -> bool {
        matches!(self, Self::Broadcast)
    }
}

/// Tracks a single peer in the context of [`TransactionsManager`].
#[derive(Debug)]
pub struct PeerMetadata<N: NetworkPrimitives = EthNetworkPrimitives> {
    /// Optimistically keeps track of transactions that we know the peer has seen. Optimistic, in
    /// the sense that transactions are preemptively marked as seen by peer when they are sent to
    /// the peer.
    seen_transactions: LruCache<TxHash, FbBuildHasher<32>>,
    /// A communication channel directly to the peer's session task.
    request_tx: PeerRequestSender<PeerRequest<N>>,
    /// negotiated version of the session.
    version: EthVersion,
    /// The peer's client version.
    client_version: Arc<str>,
    /// The kind of peer.
    peer_kind: PeerKind,
}

impl<N: NetworkPrimitives> PeerMetadata<N> {
    /// Returns a new instance of [`PeerMetadata`].
    pub fn new(
        request_tx: PeerRequestSender<PeerRequest<N>>,
        version: EthVersion,
        client_version: Arc<str>,
        max_transactions_seen_by_peer: u32,
        peer_kind: PeerKind,
    ) -> Self {
        Self {
            seen_transactions: LruCache::with_hasher(
                max_transactions_seen_by_peer,
                Default::default(),
            ),
            request_tx,
            version,
            client_version,
            peer_kind,
        }
    }

    /// Returns a reference to the peer's request sender channel.
    pub const fn request_tx(&self) -> &PeerRequestSender<PeerRequest<N>> {
        &self.request_tx
    }

    /// Returns a mutable reference to the seen transactions LRU cache.
    pub const fn seen_transactions_mut(&mut self) -> &mut LruCache<TxHash, FbBuildHasher<32>> {
        &mut self.seen_transactions
    }

    /// Returns the negotiated `EthVersion` of the session.
    pub const fn version(&self) -> EthVersion {
        self.version
    }

    /// Returns a reference to the peer's client version string.
    pub fn client_version(&self) -> &str {
        &self.client_version
    }

    /// Returns the peer's kind.
    pub const fn peer_kind(&self) -> PeerKind {
        self.peer_kind
    }
}

/// Commands to send to the [`TransactionsManager`]
#[derive(Debug)]
enum TransactionsCommand<N: NetworkPrimitives = EthNetworkPrimitives> {
    /// Propagate a transaction hash to the network.
    PropagateHash(B256),
    /// Propagate transaction hashes to a specific peer.
    PropagateHashesTo(Vec<B256>, PeerId),
    /// Request the list of active peer IDs from the [`TransactionsManager`].
    GetActivePeers(oneshot::Sender<HashSet<PeerId>>),
    /// Propagate a collection of full transactions to a specific peer.
    PropagateTransactionsTo(Vec<TxHash>, PeerId),
    /// Propagate a collection of hashes to all peers.
    PropagateTransactions(Vec<TxHash>),
    /// Propagate a collection of broadcastable transactions in full to all peers.
    BroadcastTransactions(Vec<PropagateTransaction>),
    /// Request transaction hashes known by specific peers from the [`TransactionsManager`].
    GetTransactionHashes { peers: Vec<PeerId>, tx: oneshot::Sender<HashMap<PeerId, B256Set>> },
    /// Requests a clone of the sender channel to the peer.
    GetPeerSender {
        peer_id: PeerId,
        peer_request_sender: oneshot::Sender<Option<PeerRequestSender<PeerRequest<N>>>>,
    },
}

/// All events related to transactions emitted by the network.
#[derive(Debug)]
pub enum NetworkTransactionEvent<N: NetworkPrimitives = EthNetworkPrimitives> {
    /// Represents the event of receiving a list of transactions from a peer.
    ///
    /// This indicates transactions that were broadcasted to us from the peer.
    IncomingTransactions {
        /// The ID of the peer from which the transactions were received.
        peer_id: PeerId,
        /// The received transactions.
        msg: Transactions<N::BroadcastedTransaction>,
    },
    /// Represents the event of receiving a list of transaction hashes from a peer.
    IncomingPooledTransactionHashes {
        /// The ID of the peer from which the transaction hashes were received.
        peer_id: PeerId,
        /// The received new pooled transaction hashes.
        msg: NewPooledTransactionHashes,
    },
    /// Represents the event of receiving a `GetPooledTransactions` request from a peer.
    GetPooledTransactions {
        /// The ID of the peer from which the request was received.
        peer_id: PeerId,
        /// The received `GetPooledTransactions` request.
        request: GetPooledTransactions,
        /// The sender for responding to the request with a result of `PooledTransactions`.
        response: oneshot::Sender<RequestResult<PooledTransactions<N::PooledTransaction>>>,
    },
    /// Represents the event of receiving a `GetTransactionsHandle` request.
    GetTransactionsHandle(oneshot::Sender<Option<TransactionsHandle<N>>>),
}

/// Tracks stats about the [`TransactionsManager`].
#[derive(Debug)]
pub struct PendingPoolImportsInfo {
    /// Number of transactions about to be inserted into the pool.
    pending_pool_imports: Arc<AtomicUsize>,
    /// Max number of transactions allowed to be imported concurrently.
    max_pending_pool_imports: usize,
    /// Encoded byte size of the transactions about to be inserted into the pool.
    pending_pool_imports_bytes: Arc<AtomicUsize>,
    /// Max encoded byte size of transactions allowed to be imported concurrently.
    max_pending_pool_imports_bytes: usize,
}

impl PendingPoolImportsInfo {
    /// Returns a new [`PendingPoolImportsInfo`].
    pub fn new(max_pending_pool_imports: usize, max_pending_pool_imports_bytes: usize) -> Self {
        Self {
            pending_pool_imports: Arc::new(AtomicUsize::default()),
            max_pending_pool_imports,
            pending_pool_imports_bytes: Arc::new(AtomicUsize::default()),
            max_pending_pool_imports_bytes,
        }
    }

    /// Returns `true` if the number of pool imports and their encoded byte size are both under
    /// the configured max.
    pub fn has_capacity(&self) -> bool {
        self.pending_pool_imports.load(Ordering::Relaxed) < self.max_pending_pool_imports &&
            self.pending_pool_imports_bytes.load(Ordering::Relaxed) <
                self.max_pending_pool_imports_bytes
    }
}

impl Default for PendingPoolImportsInfo {
    fn default() -> Self {
        Self::new(
            DEFAULT_MAX_COUNT_PENDING_POOL_IMPORTS,
            DEFAULT_MAX_BYTE_SIZE_PENDING_POOL_IMPORTS,
        )
    }
}

#[derive(Debug, Default)]
struct TxManagerPollDurations {
    acc_network_events: Duration,
    acc_pending_imports: Duration,
    acc_tx_events: Duration,
    acc_imported_txns: Duration,
    acc_fetch_events: Duration,
    acc_pending_fetch: Duration,
    acc_cmds: Duration,
}

impl<N: NetworkPrimitives> InMemorySize for NetworkTransactionEvent<N> {
    // `N::BroadcastedTransaction` and `N::PooledTransaction` already implement
    // `InMemorySize` via `SignedTransaction: InMemorySize`, so no extra bound is needed.
    fn size(&self) -> usize {
        match self {
            Self::IncomingTransactions { peer_id, msg } => {
                core::mem::size_of_val(peer_id) +
                    msg.0.iter().map(InMemorySize::size).sum::<usize>()
            }
            Self::IncomingPooledTransactionHashes { peer_id, msg } => {
                core::mem::size_of_val(peer_id) + msg.size()
            }
            Self::GetPooledTransactions { peer_id, request, response } => {
                core::mem::size_of_val(peer_id) +
                    request.0.len() * core::mem::size_of::<TxHash>() +
                    core::mem::size_of_val(response)
            }
            Self::GetTransactionsHandle(_) => 0,
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{
        test_utils::{
            transactions::{buffer_hash_to_tx_fetcher, new_mock_session, new_tx_manager},
            Testnet,
        },
        transactions::config::RelaxedEthAnnouncementFilter,
        NetworkConfigBuilder, NetworkManager,
    };
    use alloy_consensus::{Transaction as _, TxEip1559, TxLegacy};
    use alloy_eips::{eip2718::Encodable2718, eip4844::BlobTransactionValidationError};
    use alloy_primitives::{address, hex, Signature, TxKind, B256, U256};
    use alloy_rlp::Decodable;
    use futures::FutureExt;
    use reth_chainspec::MIN_TRANSACTION_GAS;
    use reth_ethereum_primitives::{PooledTransactionVariant, Transaction, TransactionSigned};
    use reth_network_api::{NetworkInfo, PeerKind};
    use reth_network_p2p::{
        error::{RequestError, RequestResult},
        sync::{NetworkSyncUpdater, SyncState},
    };
    use reth_storage_api::noop::NoopProvider;
    use reth_tasks::Runtime;
    use reth_transaction_pool::{
        blobstore::InMemoryBlobStore,
        error::{Eip4844PoolTransactionError, InvalidPoolTransactionError, PoolError},
        identifier::SenderIdentifiers,
        test_utils::{
            testing_pool, MockTransaction, MockTransactionFactory, OkValidator, TestPool,
            TransactionGenerator,
        },
        CoinbaseTipOrdering, EthPooledTransaction, Pool, TransactionOrigin, ValidPoolTransaction,
    };
    use secp256k1::SecretKey;
    use std::{
        future::poll_fn,
        net::{IpAddr, Ipv4Addr, SocketAddr},
        str::FromStr,
        time::Instant,
    };
    use tracing::error;

    type EthTestPool = Pool<
        OkValidator<EthPooledTransaction>,
        CoinbaseTipOrdering<EthPooledTransaction>,
        InMemoryBlobStore,
    >;

    #[tokio::test]
    async fn announcement_policy_preserves_order_and_skips_pending_and_bad_imports() {
        #[derive(Debug, Clone, Default)]
        struct RecordingPolicy(Arc<parking_lot::Mutex<Vec<(u8, TxHash, usize)>>>);

        impl AnnouncementFilteringPolicy<EthNetworkPrimitives> for RecordingPolicy {
            fn decide_on_announcement(
                &self,
                tx_type: u8,
                hash: &TxHash,
                size: usize,
            ) -> AnnouncementAcceptance {
                self.0.lock().push((tx_type, *hash, size));
                match tx_type {
                    0xff => AnnouncementAcceptance::Reject { penalize_peer: true },
                    1 => AnnouncementAcceptance::Ignore,
                    _ => AnnouncementAcceptance::Accept,
                }
            }
        }

        let (mut manager, _network) = new_tx_manager().await;
        let policy = RecordingPolicy::default();
        manager.policies =
            NetworkPolicies::new(TransactionPropagationKind::default(), policy.clone());
        let peer_id = PeerId::repeat_byte(1);
        let (peer, mut rx) = new_mock_session(peer_id, EthVersion::Eth68);
        manager.peers.insert(peer_id, peer);
        let [pending, accepted, ignored, rejected, later, bad] =
            [1, 2, 3, 4, 5, 6].map(B256::repeat_byte);
        manager.transactions_by_peers.insert(pending, Default::default());
        manager.bad_imports.insert(bad);
        manager.on_new_pooled_transaction_hashes(
            peer_id,
            NewPooledTransactionHashes68 {
                hashes: vec![pending, accepted, accepted, ignored, rejected, later, bad],
                types: vec![0xff, 2, 0xff, 1, 0xff, 2, 0xff],
                sizes: vec![100, 200, 300, 400, 500, 600, 700],
            }
            .into(),
        );

        assert_eq!(
            *policy.0.lock(),
            [(2, accepted, 200), (1, ignored, 400), (0xff, rejected, 500), (2, later, 600)]
        );
        poll_fn(|cx| {
            let _ = manager.poll_unpin(cx);
            Poll::Ready(())
        })
        .await;
        let PeerRequest::GetPooledTransactions { request, .. } = rx.try_recv().unwrap() else {
            panic!("expected transaction request")
        };
        assert_eq!(request.0, [accepted, later]);
    }

    async fn new_eth_tx_manager() -> (
        TransactionsManager<EthTestPool, EthNetworkPrimitives>,
        NetworkManager<EthNetworkPrimitives>,
    ) {
        let secret_key = SecretKey::new(&mut rand::rng());
        let client = NoopProvider::default();

        let config = NetworkConfigBuilder::new(secret_key, Runtime::test())
            .listener_port(0)
            .disable_discovery()
            .build(client);

        let pool = Pool::new(
            OkValidator::default(),
            CoinbaseTipOrdering::default(),
            InMemoryBlobStore::default(),
            Default::default(),
        );

        let transactions_manager_config = config.transactions_manager_config.clone();
        let (_network_handle, network, transactions, _) = NetworkManager::new(config)
            .await
            .unwrap()
            .into_builder()
            .transactions(pool.clone(), transactions_manager_config)
            .split_with_handle();

        (transactions, network)
    }

    fn valid_eth_pool_transaction(
        transaction: EthPooledTransaction,
    ) -> Arc<ValidPoolTransaction<EthPooledTransaction>> {
        let mut ids = SenderIdentifiers::default();
        let transaction_id =
            ids.sender_id_or_create(transaction.sender()).into_transaction_id(transaction.nonce());

        Arc::new(ValidPoolTransaction {
            propagate: false,
            transaction_id,
            transaction,
            timestamp: Instant::now(),
            origin: TransactionOrigin::External,
            authority_ids: None,
        })
    }

    fn gen_eip1559_pooled_with_nonce<R: rand::RngCore>(
        tx_gen: &mut TransactionGenerator<R>,
        nonce: u64,
    ) -> EthPooledTransaction {
        EthPooledTransaction::try_from_consensus(
            tx_gen.transaction().nonce(nonce).into_eip1559().try_into_recovered().unwrap(),
        )
        .unwrap()
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn test_ignored_tx_broadcasts_while_initially_syncing() {
        reth_tracing::init_test_tracing();
        let net = Testnet::create(3).await.spawn();
        let [peer0, peer1, _] = net.peers_array();

        let listener0 = peer0.event_listener();
        peer0.add_peer(peer1);
        let secret_key = SecretKey::new(&mut rand::rng());

        let client = NoopProvider::default();
        let pool = testing_pool();
        let config = NetworkConfigBuilder::eth(secret_key, Runtime::test())
            .disable_discovery()
            .listener_port(0)
            .build(client);
        let transactions_manager_config = config.transactions_manager_config.clone();
        let (network_handle, network, mut transactions, _) = NetworkManager::new(config)
            .await
            .unwrap()
            .into_builder()
            .transactions(pool.clone(), transactions_manager_config)
            .split_with_handle();

        tokio::task::spawn(network);

        // go to syncing (pipeline sync)
        network_handle.update_sync_state(SyncState::Syncing);
        assert!(NetworkInfo::is_syncing(&network_handle));
        assert!(NetworkInfo::is_initially_syncing(&network_handle));

        // wait for all initiator connections
        let mut established = listener0.take(2);
        while let Some(ev) = established.next().await {
            match ev {
                NetworkEvent::Peer(PeerEvent::SessionEstablished(info)) => {
                    // to insert a new peer in transactions peerset
                    transactions
                        .on_network_event(NetworkEvent::Peer(PeerEvent::SessionEstablished(info)))
                }
                NetworkEvent::Peer(PeerEvent::PeerAdded(_peer_id)) => {}
                ev => {
                    error!("unexpected event {ev:?}")
                }
            }
        }
        // random tx: <https://etherscan.io/getRawTx?tx=0x9448608d36e721ef403c53b00546068a6474d6cbab6816c3926de449898e7bce>
        let input = hex!(
            "02f871018302a90f808504890aef60826b6c94ddf4c5025d1a5742cf12f74eec246d4432c295e487e09c3bbcc12b2b80c080a0f21a4eacd0bf8fea9c5105c543be5a1d8c796516875710fafafdf16d16d8ee23a001280915021bb446d1973501a67f93d2b38894a514b976e7b46dc2fe54598d76"
        );
        let signed_tx = TransactionSigned::decode(&mut &input[..]).unwrap();
        transactions.on_network_tx_event(NetworkTransactionEvent::IncomingTransactions {
            peer_id: *peer1.peer_id(),
            msg: Transactions(vec![signed_tx.clone()]),
        });
        poll_fn(|cx| {
            let _ = transactions.poll_unpin(cx);
            Poll::Ready(())
        })
        .await;
        assert!(pool.is_empty());
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn test_tx_broadcasts_through_two_syncs() {
        reth_tracing::init_test_tracing();
        let net = Testnet::create(3).await.spawn();
        let [peer0, peer1, _] = net.peers_array();

        let listener0 = peer0.event_listener();
        peer0.add_peer(peer1);
        let secret_key = SecretKey::new(&mut rand::rng());

        let client = NoopProvider::default();
        let pool = testing_pool();
        let config = NetworkConfigBuilder::new(secret_key, Runtime::test())
            .disable_discovery()
            .listener_port(0)
            .build(client);
        let transactions_manager_config = config.transactions_manager_config.clone();
        let (network_handle, network, mut transactions, _) = NetworkManager::new(config)
            .await
            .unwrap()
            .into_builder()
            .transactions(pool.clone(), transactions_manager_config)
            .split_with_handle();

        tokio::task::spawn(network);

        // go to syncing (pipeline sync) to idle and then to syncing (live)
        network_handle.update_sync_state(SyncState::Syncing);
        assert!(NetworkInfo::is_syncing(&network_handle));
        network_handle.update_sync_state(SyncState::Idle);
        assert!(!NetworkInfo::is_syncing(&network_handle));
        network_handle.update_sync_state(SyncState::Syncing);
        assert!(NetworkInfo::is_syncing(&network_handle));

        // wait for all initiator connections
        let mut established = listener0.take(2);
        while let Some(ev) = established.next().await {
            match ev {
                NetworkEvent::ActivePeerSession { .. } |
                NetworkEvent::Peer(PeerEvent::SessionEstablished(_)) => {
                    // to insert a new peer in transactions peerset
                    transactions.on_network_event(ev);
                }
                NetworkEvent::Peer(PeerEvent::PeerAdded(_peer_id)) => {}
                _ => {
                    error!("unexpected event {ev:?}")
                }
            }
        }
        // random tx: <https://etherscan.io/getRawTx?tx=0x9448608d36e721ef403c53b00546068a6474d6cbab6816c3926de449898e7bce>
        let input = hex!(
            "02f871018302a90f808504890aef60826b6c94ddf4c5025d1a5742cf12f74eec246d4432c295e487e09c3bbcc12b2b80c080a0f21a4eacd0bf8fea9c5105c543be5a1d8c796516875710fafafdf16d16d8ee23a001280915021bb446d1973501a67f93d2b38894a514b976e7b46dc2fe54598d76"
        );
        let signed_tx = TransactionSigned::decode(&mut &input[..]).unwrap();
        transactions.on_network_tx_event(NetworkTransactionEvent::IncomingTransactions {
            peer_id: *peer1.peer_id(),
            msg: Transactions(vec![signed_tx.clone()]),
        });
        poll_fn(|cx| {
            let _ = transactions.poll_unpin(cx);
            Poll::Ready(())
        })
        .await;
        assert!(!NetworkInfo::is_initially_syncing(&network_handle));
        assert!(NetworkInfo::is_syncing(&network_handle));
        assert!(!pool.is_empty());
    }

    // Ensure that the transaction manager correctly handles the `IncomingPooledTransactionHashes`
    // event and is able to retrieve the corresponding transactions.
    #[tokio::test(flavor = "multi_thread")]
    async fn test_handle_incoming_transactions_hashes() {
        reth_tracing::init_test_tracing();

        let secret_key = SecretKey::new(&mut rand::rng());
        let client = NoopProvider::default();

        let config = NetworkConfigBuilder::new(secret_key, Runtime::test())
            // let OS choose port
            .listener_port(0)
            .disable_discovery()
            .build(client);

        let pool = testing_pool();

        let transactions_manager_config = config.transactions_manager_config.clone();
        let (_network_handle, _network, mut tx_manager, _) = NetworkManager::new(config)
            .await
            .unwrap()
            .into_builder()
            .transactions(pool.clone(), transactions_manager_config)
            .split_with_handle();

        let peer_id_1 = PeerId::repeat_byte(1);
        let eth_version = EthVersion::Eth66;

        let txs = vec![TransactionSigned::new_unhashed(
            Transaction::Legacy(TxLegacy {
                chain_id: Some(4),
                nonce: 15u64,
                gas_price: 2200000000,
                gas_limit: 34811,
                to: TxKind::Call(address!("cf7f9e66af820a19257a2108375b180b0ec49167")),
                value: U256::from(1234u64),
                input: Default::default(),
            }),
            Signature::new(
                U256::from_str(
                    "0x35b7bfeb9ad9ece2cbafaaf8e202e706b4cfaeb233f46198f00b44d4a566a981",
                )
                .unwrap(),
                U256::from_str(
                    "0x612638fb29427ca33b9a3be2a0a561beecfe0269655be160d35e72d366a6a860",
                )
                .unwrap(),
                true,
            ),
        )];

        let txs_hashes: Vec<B256> = txs.iter().map(|tx| *tx.hash()).collect();

        let (peer_1, mut to_mock_session_rx) = new_mock_session(peer_id_1, eth_version);
        tx_manager.peers.insert(peer_id_1, peer_1);

        assert!(pool.is_empty());

        tx_manager.on_network_tx_event(NetworkTransactionEvent::IncomingPooledTransactionHashes {
            peer_id: peer_id_1,
            msg: NewPooledTransactionHashes::from(NewPooledTransactionHashes66::from(
                txs_hashes.clone(),
            )),
        });

        // advance the transaction manager future to send the request
        poll_fn(|cx| {
            let _ = tx_manager.poll_unpin(cx);
            Poll::Ready(())
        })
        .await;

        // mock session of peer_1 receives request
        let req = to_mock_session_rx
            .recv()
            .await
            .expect("peer_1 session should receive request with buffered hashes");
        let PeerRequest::GetPooledTransactions { request, response } = req else { unreachable!() };
        assert_eq!(request, GetPooledTransactions::from(txs_hashes.clone()));

        let message: Vec<PooledTransactionVariant> = txs
            .into_iter()
            .map(|tx| {
                PooledTransactionVariant::try_from(tx)
                    .expect("Failed to convert MockTransaction to PooledTransaction")
            })
            .collect();

        // return the transactions corresponding to the transaction hashes.
        response
            .send(Ok(PooledTransactions(message)))
            .expect("should send peer_1 response to tx manager");

        // adance the transaction manager future
        poll_fn(|cx| {
            let _ = tx_manager.poll_unpin(cx);
            Poll::Ready(())
        })
        .await;

        // ensure that the transactions corresponding to the transaction hashes have been
        // successfully retrieved and stored in the Pool.
        assert_eq!(pool.get_all(txs_hashes.clone()).len(), txs_hashes.len());
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn test_handle_incoming_transactions() {
        reth_tracing::init_test_tracing();
        let net = Testnet::create(3).await.spawn();
        let [peer0, peer1, _] = net.peers_array();

        let listener0 = peer0.event_listener();

        peer0.add_peer(peer1);
        let secret_key = SecretKey::new(&mut rand::rng());

        let client = NoopProvider::default();
        let pool = testing_pool();
        let config = NetworkConfigBuilder::new(secret_key, Runtime::test())
            .disable_discovery()
            .listener_port(0)
            .build(client);
        let transactions_manager_config = config.transactions_manager_config.clone();
        let (network_handle, network, mut transactions, _) = NetworkManager::new(config)
            .await
            .unwrap()
            .into_builder()
            .transactions(pool.clone(), transactions_manager_config)
            .split_with_handle();
        tokio::task::spawn(network);

        network_handle.update_sync_state(SyncState::Idle);

        assert!(!NetworkInfo::is_syncing(&network_handle));

        // wait for all initiator connections
        let mut established = listener0.take(2);
        while let Some(ev) = established.next().await {
            match ev {
                NetworkEvent::ActivePeerSession { .. } |
                NetworkEvent::Peer(PeerEvent::SessionEstablished(_)) => {
                    // to insert a new peer in transactions peerset
                    transactions.on_network_event(ev);
                }
                NetworkEvent::Peer(PeerEvent::PeerAdded(_peer_id)) => {}
                ev => {
                    error!("unexpected event {ev:?}")
                }
            }
        }
        // random tx: <https://etherscan.io/getRawTx?tx=0x9448608d36e721ef403c53b00546068a6474d6cbab6816c3926de449898e7bce>
        let input = hex!(
            "02f871018302a90f808504890aef60826b6c94ddf4c5025d1a5742cf12f74eec246d4432c295e487e09c3bbcc12b2b80c080a0f21a4eacd0bf8fea9c5105c543be5a1d8c796516875710fafafdf16d16d8ee23a001280915021bb446d1973501a67f93d2b38894a514b976e7b46dc2fe54598d76"
        );
        let signed_tx = TransactionSigned::decode(&mut &input[..]).unwrap();
        transactions.on_network_tx_event(NetworkTransactionEvent::IncomingTransactions {
            peer_id: *peer1.peer_id(),
            msg: Transactions(vec![signed_tx.clone()]),
        });
        assert!(transactions
            .transactions_by_peers
            .get(signed_tx.tx_hash())
            .unwrap()
            .contains(peer1.peer_id()));

        // advance the transaction manager future
        poll_fn(|cx| {
            let _ = transactions.poll_unpin(cx);
            Poll::Ready(())
        })
        .await;

        assert!(!pool.is_empty());
        assert!(pool.get(signed_tx.tx_hash()).is_some());
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn test_session_closed_cleans_transaction_peer_state() {
        let (mut tx_manager, _network) = new_tx_manager().await;
        let peer_id = PeerId::repeat_byte(1);
        let fallback_peer = PeerId::repeat_byte(2);
        let (peer, _) = new_mock_session(peer_id, EthVersion::Eth66);
        let (fallback, _) = new_mock_session(fallback_peer, EthVersion::Eth66);
        let hash_shared = B256::from_slice(&[1; 32]);

        tx_manager.peers.insert(peer_id, peer);
        tx_manager.peers.insert(fallback_peer, fallback);
        buffer_hash_to_tx_fetcher(&mut tx_manager.transaction_fetcher, hash_shared, peer_id, None);
        buffer_hash_to_tx_fetcher(
            &mut tx_manager.transaction_fetcher,
            hash_shared,
            fallback_peer,
            None,
        );
        assert_eq!(
            tx_manager.transaction_fetcher.candidate_peers(&hash_shared),
            vec![peer_id, fallback_peer]
        );

        tx_manager.on_network_event(NetworkEvent::Peer(PeerEvent::SessionClosed {
            peer_id,
            reason: None,
        }));

        // peer removed from peers map and from the fetcher
        assert!(!tx_manager.peers.contains_key(&peer_id));
        assert!(tx_manager.transaction_fetcher.queued_hashes(&peer_id).is_empty());
        // fallback peer is still available for the hash
        assert_eq!(
            tx_manager.transaction_fetcher.candidate_peers(&hash_shared),
            vec![fallback_peer]
        );
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn test_bad_blob_sidecar_not_cached_as_bad_import() {
        let (mut tx_manager, _network) = new_tx_manager().await;
        let peer_id = PeerId::repeat_byte(1);
        let hash = B256::from_slice(&[1; 32]);

        tx_manager.network.update_sync_state(SyncState::Idle);
        tx_manager.transactions_by_peers.insert(hash, smallvec::smallvec![peer_id]);

        let err = PoolError::new(
            hash,
            InvalidPoolTransactionError::Eip4844(Eip4844PoolTransactionError::InvalidEip4844Blob(
                BlobTransactionValidationError::InvalidProof,
            )),
        );

        tx_manager.on_bad_import(err);

        assert!(!tx_manager.bad_imports.contains(&hash));
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn test_missing_blob_sidecar_not_cached_as_bad_import() {
        let (mut tx_manager, _network) = new_tx_manager().await;
        let peer_id = PeerId::repeat_byte(1);
        let hash = B256::from_slice(&[3; 32]);

        tx_manager.network.update_sync_state(SyncState::Idle);
        tx_manager.transactions_by_peers.insert(hash, smallvec::smallvec![peer_id]);

        let err = PoolError::new(
            hash,
            InvalidPoolTransactionError::Eip4844(
                Eip4844PoolTransactionError::MissingEip4844BlobSidecar,
            ),
        );

        tx_manager.on_bad_import(err);

        assert!(!tx_manager.bad_imports.contains(&hash));
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn test_non_blob_sidecar_error_still_cached_as_bad_import() {
        let (mut tx_manager, _network) = new_tx_manager().await;
        let peer_id = PeerId::repeat_byte(1);
        let hash = B256::from_slice(&[2; 32]);

        tx_manager.network.update_sync_state(SyncState::Idle);
        tx_manager.transactions_by_peers.insert(hash, smallvec::smallvec![peer_id]);

        let err = PoolError::new(
            hash,
            InvalidPoolTransactionError::Eip4844(Eip4844PoolTransactionError::NoEip4844Blobs),
        );

        tx_manager.on_bad_import(err);

        assert!(tx_manager.bad_imports.contains(&hash));
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn test_on_get_pooled_transactions_network() {
        reth_tracing::init_test_tracing();
        let net = Testnet::create(2).await.spawn();
        let [peer0, peer1] = net.peers_array();

        let listener0 = peer0.event_listener();

        peer0.add_peer(peer1);
        let secret_key = SecretKey::new(&mut rand::rng());

        let client = NoopProvider::default();
        let pool = testing_pool();
        let config = NetworkConfigBuilder::new(secret_key, Runtime::test())
            .disable_discovery()
            .listener_port(0)
            .build(client);
        let transactions_manager_config = config.transactions_manager_config.clone();
        let (network_handle, network, mut transactions, _) = NetworkManager::new(config)
            .await
            .unwrap()
            .into_builder()
            .transactions(pool.clone(), transactions_manager_config)
            .split_with_handle();
        tokio::task::spawn(network);

        network_handle.update_sync_state(SyncState::Idle);

        assert!(!NetworkInfo::is_syncing(&network_handle));

        // wait for all initiator connections
        let mut established = listener0.take(2);
        while let Some(ev) = established.next().await {
            match ev {
                NetworkEvent::ActivePeerSession { .. } |
                NetworkEvent::Peer(PeerEvent::SessionEstablished(_)) => {
                    transactions.on_network_event(ev);
                }
                NetworkEvent::Peer(PeerEvent::PeerAdded(_peer_id)) => {}
                ev => {
                    error!("unexpected event {ev:?}")
                }
            }
        }

        let tx = MockTransaction::eip1559();
        let _ = transactions
            .pool
            .add_transaction(reth_transaction_pool::TransactionOrigin::External, tx.clone())
            .await;

        let request = GetPooledTransactions(vec![*tx.get_hash()]);

        let (send, receive) =
            oneshot::channel::<RequestResult<PooledTransactions<PooledTransactionVariant>>>();

        transactions.on_network_tx_event(NetworkTransactionEvent::GetPooledTransactions {
            peer_id: *peer1.peer_id(),
            request,
            response: send,
        });

        match receive.await.unwrap() {
            Ok(PooledTransactions(transactions)) => {
                assert_eq!(transactions.len(), 1);
            }
            Err(e) => {
                panic!("error: {e:?}");
            }
        }
    }

    // Ensure that when the remote peer only returns part of the requested transactions, the
    // replied transactions are removed from the `tx_fetcher`, while the unresponsive ones are
    // re-buffered.
    #[tokio::test]
    async fn test_partially_tx_response() {
        reth_tracing::init_test_tracing();

        let mut tx_manager = new_tx_manager().await.0;
        let tx_fetcher = &mut tx_manager.transaction_fetcher;

        let peer_id_1 = PeerId::repeat_byte(1);
        let eth_version = EthVersion::Eth66;

        let txs = vec![
            TransactionSigned::new_unhashed(
                Transaction::Legacy(TxLegacy {
                    chain_id: Some(4),
                    nonce: 15u64,
                    gas_price: 2200000000,
                    gas_limit: 34811,
                    to: TxKind::Call(address!("cf7f9e66af820a19257a2108375b180b0ec49167")),
                    value: U256::from(1234u64),
                    input: Default::default(),
                }),
                Signature::new(
                    U256::from_str(
                        "0x35b7bfeb9ad9ece2cbafaaf8e202e706b4cfaeb233f46198f00b44d4a566a981",
                    )
                    .unwrap(),
                    U256::from_str(
                        "0x612638fb29427ca33b9a3be2a0a561beecfe0269655be160d35e72d366a6a860",
                    )
                    .unwrap(),
                    true,
                ),
            ),
            TransactionSigned::new_unhashed(
                Transaction::Eip1559(TxEip1559 {
                    chain_id: 4,
                    nonce: 26u64,
                    max_priority_fee_per_gas: 1500000000,
                    max_fee_per_gas: 1500000013,
                    gas_limit: MIN_TRANSACTION_GAS,
                    to: TxKind::Call(address!("61815774383099e24810ab832a5b2a5425c154d5")),
                    value: U256::from(3000000000000000000u64),
                    input: Default::default(),
                    access_list: Default::default(),
                }),
                Signature::new(
                    U256::from_str(
                        "0x59e6b67f48fb32e7e570dfb11e042b5ad2e55e3ce3ce9cd989c7e06e07feeafd",
                    )
                    .unwrap(),
                    U256::from_str(
                        "0x016b83f4f980694ed2eee4d10667242b1f40dc406901b34125b008d334d47469",
                    )
                    .unwrap(),
                    true,
                ),
            ),
        ];

        let txs_hashes: Vec<B256> = txs.iter().map(|tx| *tx.hash()).collect();

        let (mut peer_1, mut to_mock_session_rx) = new_mock_session(peer_id_1, eth_version);
        // mark hashes as seen by peer so it can fish them out from the cache for hashes pending
        // fetch
        peer_1.seen_transactions.insert(txs_hashes[0]);
        peer_1.seen_transactions.insert(txs_hashes[1]);
        tx_manager.peers.insert(peer_id_1, peer_1);

        buffer_hash_to_tx_fetcher(tx_fetcher, txs_hashes[0], peer_id_1, None);
        buffer_hash_to_tx_fetcher(tx_fetcher, txs_hashes[1], peer_id_1, None);

        // peer_1 is idle
        assert!(tx_fetcher.is_idle(&peer_id_1));
        assert_eq!(tx_fetcher.num_inflight_requests(), 0);

        // sends requests for buffered hashes to peer_1
        assert_eq!(tx_fetcher.dispatch(&tx_manager.peers, usize::MAX), 1);

        assert_eq!(tx_fetcher.num_pending_hashes(), 0);
        // as long as request is in flight peer_1 is not idle
        assert!(!tx_fetcher.is_idle(&peer_id_1));
        assert_eq!(tx_fetcher.num_inflight_requests(), 1);

        // mock session of peer_1 receives request
        let req = to_mock_session_rx
            .recv()
            .await
            .expect("peer_1 session should receive request with buffered hashes");
        let PeerRequest::GetPooledTransactions { response, .. } = req else { unreachable!() };

        let message: Vec<PooledTransactionVariant> = txs
            .into_iter()
            .take(1)
            .map(|tx| {
                PooledTransactionVariant::try_from(tx)
                    .expect("Failed to convert MockTransaction to PooledTransaction")
            })
            .collect();
        // response partial request
        response
            .send(Ok(PooledTransactions(message)))
            .expect("should send peer_1 response to tx manager");
        let Some(FetchEvent::TransactionsFetched { peer_id, .. }) = tx_fetcher.next().await else {
            unreachable!()
        };

        // request has resolved, peer_1 is idle again
        assert!(tx_fetcher.is_idle(&peer_id));
        assert_eq!(tx_fetcher.num_inflight_requests(), 0);
        // the undelivered hash at the end of the request is retried with peer_1
        assert_eq!(tx_fetcher.num_pending_hashes(), 1);
        assert_eq!(tx_fetcher.candidate_peers(&txs_hashes[1]), vec![peer_id_1]);
    }

    /// Tests that hashes of a failed request are retried with the alternate peer that announced
    /// them, and are given up on when no peer is left to fetch them from.
    #[tokio::test]
    async fn test_failed_request_retries_on_alternate_peer() {
        reth_tracing::init_test_tracing();

        let mut tx_manager = new_tx_manager().await.0;
        let tx_fetcher = &mut tx_manager.transaction_fetcher;

        let peer_id_1 = PeerId::repeat_byte(1);
        let peer_id_2 = PeerId::repeat_byte(2);
        let eth_version = EthVersion::Eth66;
        let seen_hashes = [B256::from_slice(&[1; 32]), B256::from_slice(&[2; 32])];

        let (peer_1, mut to_mock_session_rx_1) = new_mock_session(peer_id_1, eth_version);
        let (peer_2, mut to_mock_session_rx_2) = new_mock_session(peer_id_2, eth_version);
        tx_manager.peers.insert(peer_id_1, peer_1);
        tx_manager.peers.insert(peer_id_2, peer_2);

        // both peers announce the hashes, peer_1 first
        for hash in seen_hashes {
            buffer_hash_to_tx_fetcher(tx_fetcher, hash, peer_id_1, None);
            buffer_hash_to_tx_fetcher(tx_fetcher, hash, peer_id_2, None);
        }
        assert!(tx_fetcher.is_idle(&peer_id_1));
        assert!(tx_fetcher.is_idle(&peer_id_2));

        // the hashes are requested from peer_1 only
        assert_eq!(tx_fetcher.dispatch(&tx_manager.peers, usize::MAX), 1);
        assert_eq!(tx_fetcher.num_pending_hashes(), 0);
        assert!(!tx_fetcher.is_idle(&peer_id_1));
        assert!(tx_fetcher.is_idle(&peer_id_2));

        let req = to_mock_session_rx_1
            .recv()
            .await
            .expect("peer_1 session should receive request with buffered hashes");
        let PeerRequest::GetPooledTransactions { request, response } = req else { unreachable!() };
        let GetPooledTransactions(hashes) = request;
        assert_eq!(
            hashes.into_iter().collect::<B256Set>(),
            seen_hashes.into_iter().collect::<B256Set>()
        );

        // fail request to peer_1
        response
            .send(Err(RequestError::BadResponse))
            .expect("should send peer_1 response to tx manager");
        let Some(FetchEvent::FetchError { peer_id, .. }) = tx_fetcher.next().await else {
            unreachable!()
        };
        assert_eq!(peer_id, peer_id_1);

        // request has resolved, peer_1 is idle again and the hashes are pending for peer_2 only
        assert!(tx_fetcher.is_idle(&peer_id_1));
        assert_eq!(tx_fetcher.num_inflight_requests(), 0);
        assert_eq!(tx_fetcher.num_pending_hashes(), 2);
        assert_eq!(tx_fetcher.candidate_peers(&seen_hashes[0]), vec![peer_id_2]);

        // the hashes are retried with peer_2
        assert_eq!(tx_fetcher.dispatch(&tx_manager.peers, usize::MAX), 1);
        assert_eq!(tx_fetcher.num_pending_hashes(), 0);
        let req = to_mock_session_rx_2
            .recv()
            .await
            .expect("peer_2 session should receive request with buffered hashes");
        let PeerRequest::GetPooledTransactions { response, .. } = req else { unreachable!() };

        // fail request to peer_2 as well
        response
            .send(Err(RequestError::BadResponse))
            .expect("should send peer_2 response to tx manager");
        let Some(FetchEvent::FetchError { .. }) = tx_fetcher.next().await else { unreachable!() };

        // no peer is left to fetch the hashes from, they are dropped
        assert_eq!(tx_fetcher.num_hashes(), 0);
        assert_eq!(tx_fetcher.num_inflight_requests(), 0);
    }

    #[test]
    fn test_direct_propagation_transaction_uses_2718_size() {
        let mut tx_gen = TransactionGenerator::new(rand::rng());
        let tx = tx_gen.gen_eip1559();
        let expected_size = tx.encode_2718_len();

        let tx = PropagateTransaction::new(tx);

        assert_eq!(tx.propagation_size(), expected_size);
    }

    #[test]
    fn test_transaction_builder_empty() {
        let mut builder = PropagateTransactionsBuilder::pooled(EthVersion::Eth68, 0);
        assert!(builder.is_empty());

        let mut tx_gen = TransactionGenerator::new(rand::rng());
        let tx =
            PropagateTransaction::pool_tx(valid_eth_pool_transaction(tx_gen.gen_eip1559_pooled()));
        builder.push(&tx);
        assert!(!builder.is_empty());

        let txs = builder.build();
        assert!(txs.full.is_none());
        let txs = txs.pooled.unwrap();
        assert_eq!(txs.len(), 1);
    }

    #[test]
    fn test_pooled_propagation_transaction_encoder_length_matches_network_encoding() {
        let mut tx_gen = TransactionGenerator::new(rand::rng());
        let tx = valid_eth_pool_transaction(tx_gen.gen_eip1559_pooled());
        let pooled = PropagatePooledTransactionEncoder::new(tx);

        let mut pooled_encoded = Vec::new();
        pooled.encode(&mut pooled_encoded);
        assert_eq!(pooled.length(), pooled_encoded.len());

        let broadcast = BroadcastPoolTransactions(vec![LazyEncoded::new(pooled)]);
        let mut first_encoded = Vec::new();
        broadcast.encode(&mut first_encoded);
        let mut second_encoded = Vec::new();
        broadcast.encode(&mut second_encoded);
        assert_eq!(first_encoded, second_encoded);

        let mut encoded = first_encoded.as_slice();
        let decoded = Transactions::<TransactionSigned>::decode(&mut encoded).unwrap();
        assert_eq!(decoded.len(), 1);
        assert!(encoded.is_empty());
    }

    #[test]
    fn test_transaction_builder_large() {
        let mut builder = PropagateTransactionsBuilder::full(EthVersion::Eth68, 0);
        assert!(builder.is_empty());

        let mut tx_gen = TransactionGenerator::new(rand::rng());
        let mut tx = tx_gen.gen_eip1559_pooled();
        // create a transaction that still fits
        tx.encoded_length = DEFAULT_SOFT_LIMIT_BYTE_SIZE_TRANSACTIONS_BROADCAST_MESSAGE + 1;
        let tx = PropagateTransaction::pool_tx(valid_eth_pool_transaction(tx));
        builder.push(&tx);
        assert!(!builder.is_empty());

        let txs = builder.clone().build();
        assert!(txs.pooled.is_none());
        let txs = txs.full.unwrap();
        assert_eq!(txs.len(), 1);

        builder.push(&tx);

        let txs = builder.clone().build();
        let pooled = txs.pooled.unwrap();
        assert_eq!(pooled.len(), 1);
        let txs = txs.full.unwrap();
        assert_eq!(txs.len(), 1);
    }

    #[test]
    fn test_transaction_builder_eip4844() {
        let mut builder = PropagateTransactionsBuilder::full(EthVersion::Eth68, 0);
        assert!(builder.is_empty());

        let mut tx_gen = TransactionGenerator::new(rand::rng());
        let tx =
            PropagateTransaction::pool_tx(valid_eth_pool_transaction(tx_gen.gen_eip4844_pooled()));
        builder.push(&tx);
        assert!(!builder.is_empty());

        let txs = builder.clone().build();
        assert!(txs.full.is_none());
        let txs = txs.pooled.unwrap();
        assert_eq!(txs.len(), 1);

        let tx =
            PropagateTransaction::pool_tx(valid_eth_pool_transaction(tx_gen.gen_eip1559_pooled()));
        builder.push(&tx);

        let txs = builder.clone().build();
        let pooled = txs.pooled.unwrap();
        assert_eq!(pooled.len(), 1);
        let txs = txs.full.unwrap();
        assert_eq!(txs.len(), 1);
    }

    #[tokio::test]
    async fn test_propagate_full() {
        reth_tracing::init_test_tracing();

        let (mut tx_manager, network) = new_eth_tx_manager().await;
        let peer_id = PeerId::random();

        // ensure not syncing
        network.handle().update_sync_state(SyncState::Idle);

        // mock a peer
        let (tx, _rx) = mpsc::channel::<PeerRequest>(1);

        let session_info = SessionInfo {
            peer_id,
            remote_addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0),
            client_version: Arc::from(""),
            capabilities: Arc::new(vec![].into()),
            status: Arc::new(Default::default()),
            version: EthVersion::Eth68,
            peer_kind: PeerKind::Basic,
        };
        let messages: PeerRequestSender<PeerRequest> = PeerRequestSender::new(peer_id, tx);
        tx_manager
            .on_network_event(NetworkEvent::ActivePeerSession { info: session_info, messages });
        let mut propagate = vec![];
        let mut tx_gen = TransactionGenerator::new(rand::rng());
        let eip1559_tx = valid_eth_pool_transaction(tx_gen.gen_eip1559_pooled());
        propagate.push(eip1559_tx.clone());
        let eip4844_tx = valid_eth_pool_transaction(tx_gen.gen_eip4844_pooled());
        propagate.push(eip4844_tx.clone());

        let propagated = tx_manager.propagate_transactions(
            propagate.clone().into_iter().map(PropagateTransaction::pool_tx).collect(),
            PropagationMode::Basic,
        );
        assert_eq!(propagated.len(), 2);
        let prop_txs = propagated.get(eip1559_tx.transaction.hash()).unwrap();
        assert_eq!(prop_txs.len(), 1);
        assert!(prop_txs[0].is_full());

        let prop_txs = propagated.get(eip4844_tx.transaction.hash()).unwrap();
        assert_eq!(prop_txs.len(), 1);
        assert!(prop_txs[0].is_hash());

        let peer = tx_manager.peers.get(&peer_id).unwrap();
        assert!(peer.seen_transactions.contains(eip1559_tx.transaction.hash()));
        assert!(peer.seen_transactions.contains(eip1559_tx.transaction.hash()));
        peer.seen_transactions.contains(eip4844_tx.transaction.hash());

        // propagate again
        let propagated = tx_manager.propagate_transactions(
            propagate.into_iter().map(PropagateTransaction::pool_tx).collect(),
            PropagationMode::Basic,
        );
        assert!(propagated.is_empty());
    }

    #[test]
    fn test_eth72_hashes_builder_sets_cell_mask_for_blob_txs() {
        let mut tx_gen = TransactionGenerator::new(rand::rng());

        // no blob transactions: the mask stays unset and encodes as a zero mask
        let mut builder = PooledTransactionsHashesBuilder::new(EthVersion::Eth72);
        builder.push(&PropagateTransaction::pool_tx(valid_eth_pool_transaction(
            tx_gen.gen_eip1559_pooled(),
        )));
        let msg = builder.build();
        assert_eq!(msg.as_eth72().unwrap().cell_mask, None);

        // announcing a blob transaction advertises every cell as available
        let mut builder = PooledTransactionsHashesBuilder::new(EthVersion::Eth72);
        builder.push(&PropagateTransaction::pool_tx(valid_eth_pool_transaction(
            tx_gen.gen_eip1559_pooled(),
        )));
        builder.push_pooled(valid_eth_pool_transaction(tx_gen.gen_eip4844_pooled()));
        let msg = builder.build();
        assert_eq!(
            msg.as_eth72().unwrap().cell_mask,
            Some(NewPooledTransactionHashes72::ALL_CELLS_MASK)
        );
    }

    #[tokio::test]
    async fn test_truncated_hash_announcement_not_marked_seen() {
        reth_tracing::init_test_tracing();

        let (mut tx_manager, network) = new_eth_tx_manager().await;
        // all peers receive hash announcements only
        tx_manager.config.propagation_mode = TransactionPropagationMode::Max(0);

        // ensure not syncing
        network.handle().update_sync_state(SyncState::Idle);

        let peer_id = PeerId::random();
        let (peer, _rx) = new_mock_session(peer_id, EthVersion::Eth68);
        tx_manager.peers.insert(peer_id, peer);

        // one more transaction than fits into a single hashes broadcast message
        let mut tx_gen = TransactionGenerator::new(rand::rng());
        let txs = (0..=SOFT_LIMIT_COUNT_HASHES_IN_NEW_POOLED_TRANSACTIONS_BROADCAST_MESSAGE)
            .map(|nonce| {
                valid_eth_pool_transaction(gen_eip1559_pooled_with_nonce(&mut tx_gen, nonce as u64))
            })
            .collect::<Vec<_>>();
        let last_sent = *txs[txs.len() - 2].hash();
        let truncated = *txs[txs.len() - 1].hash();

        let propagated = tx_manager.propagate_transactions(
            txs.into_iter().map(PropagateTransaction::pool_tx).collect(),
            PropagationMode::Basic,
        );

        // the truncated hash was not sent, so it must not be tracked as seen by the peer
        assert!(propagated.get(&truncated).is_none());
        let peer = tx_manager.peers.get(&peer_id).unwrap();
        assert!(!peer.seen_transactions.contains(&truncated));
        assert!(peer.seen_transactions.contains(&last_sent));
    }

    #[tokio::test]
    async fn test_propagate_pending_txs_while_initially_syncing() {
        reth_tracing::init_test_tracing();

        let (mut tx_manager, network) = new_eth_tx_manager().await;
        let peer_id = PeerId::random();

        // Keep the node in initial sync mode.
        network.handle().update_sync_state(SyncState::Syncing);
        assert!(NetworkInfo::is_initially_syncing(&network.handle()));

        // Add a peer so propagation has a destination.
        let (peer, _rx) = new_mock_session(peer_id, EthVersion::Eth68);
        tx_manager.peers.insert(peer_id, peer);

        let mut tx_gen = TransactionGenerator::new(rand::rng());
        let tx = gen_eip1559_pooled_with_nonce(&mut tx_gen, 0);
        let tx_hash = *tx.hash();
        tx_manager
            .pool
            .add_transaction(reth_transaction_pool::TransactionOrigin::External, tx.clone())
            .await
            .expect("transaction should be accepted into the pool");

        tx_manager.on_new_pending_transactions(vec![tx_hash]);

        let peer = tx_manager.peers.get(&peer_id).expect("peer should exist");
        assert!(peer.seen_transactions.contains(&tx_hash));
    }

    #[tokio::test]
    async fn test_relaxed_filter_ignores_unknown_tx_types() {
        reth_tracing::init_test_tracing();

        let transactions_manager_config = TransactionsManagerConfig::default();

        let propagation_policy = TransactionPropagationKind::default();
        let announcement_policy = RelaxedEthAnnouncementFilter::default();

        let policy_bundle = NetworkPolicies::new(propagation_policy, announcement_policy);

        let pool = testing_pool();
        let secret_key = SecretKey::new(&mut rand::rng());
        let client = NoopProvider::default();

        let network_config = NetworkConfigBuilder::new(secret_key, Runtime::test())
            .listener_port(0)
            .disable_discovery()
            .build(client.clone());

        let mut network_manager = NetworkManager::new(network_config).await.unwrap();
        let (to_tx_manager_tx, from_network_rx) =
            reth_metrics::common::mpsc::memory_bounded_channel::<
                NetworkTransactionEvent<EthNetworkPrimitives>,
            >(
                crate::transactions::constants::tx_manager::DEFAULT_TX_MANAGER_CHANNEL_MEMORY_LIMIT_BYTES,
                "test_tx_channel",
            );
        network_manager.set_transactions(to_tx_manager_tx);
        let network_handle = network_manager.handle().clone();
        let network_service_handle = tokio::spawn(network_manager);

        let mut tx_manager = TransactionsManager::<TestPool, EthNetworkPrimitives>::with_policy(
            network_handle.clone(),
            pool.clone(),
            from_network_rx,
            transactions_manager_config,
            policy_bundle,
        );

        let peer_id = PeerId::random();
        let eth_version = EthVersion::Eth68;
        let (mock_peer_metadata, mut mock_session_rx) = new_mock_session(peer_id, eth_version);
        tx_manager.peers.insert(peer_id, mock_peer_metadata);

        let mut tx_factory = MockTransactionFactory::default();

        let valid_known_tx = tx_factory.create_eip1559();
        let known_tx_signed: Arc<ValidPoolTransaction<MockTransaction>> = Arc::new(valid_known_tx);

        let known_tx_hash = *known_tx_signed.hash();
        let known_tx_type_byte = known_tx_signed.transaction.tx_type();
        let known_tx_size = known_tx_signed.encoded_length();

        let unknown_tx_hash = B256::random();
        let unknown_tx_type_byte = 0xff_u8;
        let unknown_tx_size = 150;

        let announcement_msg = NewPooledTransactionHashes::Eth68(NewPooledTransactionHashes68 {
            types: vec![known_tx_type_byte, unknown_tx_type_byte],
            sizes: vec![known_tx_size, unknown_tx_size],
            hashes: vec![known_tx_hash, unknown_tx_hash],
        });

        tx_manager.on_new_pooled_transaction_hashes(peer_id, announcement_msg);

        poll_fn(|cx| {
            let _ = tx_manager.poll_unpin(cx);
            Poll::Ready(())
        })
        .await;

        let mut requested_hashes_in_getpooled = B256Set::default();
        let mut unexpected_request_received = false;

        match tokio::time::timeout(std::time::Duration::from_millis(200), mock_session_rx.recv())
            .await
        {
            Ok(Some(PeerRequest::GetPooledTransactions { request, response: tx_response_ch })) => {
                let GetPooledTransactions(hashes) = request;
                for hash in hashes {
                    requested_hashes_in_getpooled.insert(hash);
                }
                let _ = tx_response_ch.send(Ok(PooledTransactions(vec![])));
            }
            Ok(Some(other_request)) => {
                tracing::error!(?other_request, "Received unexpected PeerRequest type");
                unexpected_request_received = true;
            }
            Ok(None) => tracing::info!("Mock session channel closed or no request received."),
            Err(_timeout_err) => {
                tracing::info!("Timeout: No GetPooledTransactions request received.")
            }
        }

        assert!(
            requested_hashes_in_getpooled.contains(&known_tx_hash),
            "Should have requested the known EIP-1559 transaction. Requested: {requested_hashes_in_getpooled:?}"
        );
        assert!(
            !requested_hashes_in_getpooled.contains(&unknown_tx_hash),
            "Should NOT have requested the unknown transaction type. Requested: {requested_hashes_in_getpooled:?}"
        );
        assert!(
            !unexpected_request_received,
            "An unexpected P2P request was received by the mock peer."
        );

        network_service_handle.abort();
    }
}