zakura 1.7.0

//! Zakura mempool.
//!
//! A zakurad application component that manages the active collection, reception,
//! gossip, verification, in-memory storage, eviction, and rejection of unmined Zcash
//! transactions (those that have not been confirmed in a mined block on the
//! blockchain).
//!
//! Major parts of the mempool include:
//!  * [Mempool Service][`Mempool`]
//!    * activates when the syncer is near the chain tip
//!    * spawns [download and verify tasks][`downloads::Downloads`] for each crawled or gossiped transaction
//!    * handles in-memory [storage][`storage::Storage`] of unmined transactions
//!  * [Crawler][`crawler::Crawler`]
//!    * runs in the background to periodically poll peers for fresh unmined transactions
//!  * [Queue Checker][`queue_checker::QueueChecker`]
//!    * runs in the background, polling the mempool to store newly verified transactions
//!  * [Transaction Gossip Task][`gossip::run_mempool_transaction_id_gossip`]
//!    * runs in the background and gossips newly added mempool transactions
//!      to peers

use std::{
    collections::HashSet,
    future::Future,
    iter,
    net::IpAddr,
    pin::{pin, Pin},
    task::{Context, Poll},
    time::{Duration, Instant},
};

use futures::{future::FutureExt, stream::Stream};
use tokio::sync::{broadcast, mpsc, oneshot};
use tower::{buffer::Buffer, timeout::Timeout, util::BoxCloneService, Service, ServiceExt};

use zakura_chain::{
    block::{self, Height},
    chain_sync_status::ChainSyncStatus,
    chain_tip::ChainTip,
    parameters::{Network, NetworkUpgrade},
    transaction::UnminedTxId,
};
use zakura_consensus::{error::TransactionError, transaction};
use zakura_network::{self as zn, PeerSocketAddr};
use zakura_node_services::mempool::{
    CreatedOrSpent, Gossip, MempoolChange, MempoolDisabledError, MempoolTxSubscriber, QueueSource,
    Request, Response,
};
use zakura_node_services::task::spawn_erased;
use zakura_state as zs;
use zakura_state::{ChainTipChange, TipAction};

use crate::components::sync::SyncStatus;

pub mod config;
mod crawler;
pub mod downloads;
mod error;
pub mod gossip;
mod peer_cooldown;
mod pending_outputs;
mod queue_checker;
mod storage;

/// Supplies named, timed storage operations to a benchmark harness.
#[cfg(feature = "mempool-bench")]
pub fn mempool_eviction_benchmarks(run: impl FnMut(&str, &mut dyn FnMut() -> std::time::Duration)) {
    storage::mempool_eviction_benchmarks(run);
}

#[cfg(test)]
mod tests;

pub use crate::BoxError;

pub use config::Config;
pub use crawler::Crawler;
pub use error::MempoolError;
pub(crate) use gossip::run_mempool_transaction_id_gossip;
pub use queue_checker::QueueChecker;
pub use storage::{
    ExactTipRejectionError, SameEffectsChainRejectionError, SameEffectsTipRejectionError, Storage,
};

#[cfg(test)]
pub use self::tests::UnboxMempoolError;

use downloads::{
    Downloads as TxDownloads, TransactionDownloadVerifyError, TRANSACTION_DOWNLOAD_TIMEOUT,
    TRANSACTION_VERIFY_TIMEOUT,
};

fn legacy_peer_log_label(addr: PeerSocketAddr, expose_peer_addresses: bool) -> String {
    if expose_peer_addresses {
        format!("legacy:{}", addr.remove_socket_addr_privacy())
    } else {
        "legacy:redacted".to_string()
    }
}

fn queue_source_log_label(source: &QueueSource, expose_peer_addresses: bool) -> String {
    match source {
        QueueSource::LegacySocket(addr) => {
            legacy_peer_log_label(PeerSocketAddr::from(*addr), expose_peer_addresses)
        }
        QueueSource::Zakura(peer_id) => format!("zakura:{peer_id:?}"),
    }
}

fn transaction_error_peer_log_label(
    error: &TransactionDownloadVerifyError,
    expose_peer_addresses: bool,
) -> Option<String> {
    match error {
        TransactionDownloadVerifyError::Invalid {
            advertiser_addr: Some(addr),
            ..
        } => Some(legacy_peer_log_label(*addr, expose_peer_addresses)),
        _ => None,
    }
}

type Outbound =
    Buffer<zn::Request, futures::future::BoxFuture<'static, Result<zn::Response, zn::BoxError>>>;
type State =
    Buffer<zs::Request, futures::future::BoxFuture<'static, Result<zs::Response, zs::BoxError>>>;
type ReadState = BoxCloneService<zs::ReadRequest, zs::ReadResponse, zs::BoxError>;
type TxVerifier = Buffer<
    transaction::Request,
    futures::future::BoxFuture<'static, Result<transaction::Response, TransactionError>>,
>;
type InboundTxDownloads = TxDownloads<Timeout<Outbound>, Timeout<TxVerifier>, ReadState>;

/// The maximum estimated distance to the network tip, in blocks, at which the
/// mempool and its crawler activate, and at which the mempool starts peer
/// transaction cooldowns.
///
/// This matches `getblocktemplate`'s `MAX_ESTIMATED_DISTANCE_TO_NETWORK_CHAIN_TIP`.
/// An active mempool only disables beyond [`zs::MAX_BLOCK_REORG_HEIGHT`], so
/// the gap between the two thresholds keeps the mempool from flapping.
const MAX_ESTIMATED_DISTANCE_TO_ENABLE: block::HeightDiff = 100;

/// Returns true when the local-clock estimate puts the best chain tip within
/// [`MAX_ESTIMATED_DISTANCE_TO_ENABLE`] blocks of the network tip.
/// Test networks bypass the estimate because mining can stop for long periods.
pub(crate) fn is_estimated_close_to_network_tip(chain_tip_change: &ChainTipChange) -> bool {
    if chain_tip_change.network().is_a_test_network() {
        return chain_tip_change.best_tip_height().is_some();
    }

    chain_tip_change
        .estimate_distance_to_network_chain_tip()
        .is_some_and(|(distance, _height)| distance <= MAX_ESTIMATED_DISTANCE_TO_ENABLE)
}

/// Returns the peer to ban for `error`, if the failure does not depend on this
/// node's chain tip.
///
/// [`peer_action`] decides which failures ban.
fn transaction_ban_peer(error: &TransactionDownloadVerifyError) -> Option<PeerSocketAddr> {
    let TransactionDownloadVerifyError::Invalid {
        error,
        advertiser_addr: Some(advertiser_addr),
        transaction_version,
        ..
    } = error
    else {
        return None;
    };

    let bans = match peer_action(error) {
        PeerAction::Ban => true,
        PeerAction::BanV5AndLater => *transaction_version >= 5,
        PeerAction::Cooldown | PeerAction::Ignore => false,
    };

    bans.then_some(*advertiser_addr)
}

/// The number of blocks after a network upgrade activates during which a
/// transaction built for the previous upgrade starts no cooldown.
///
/// The transaction verifier's NU6.3 branch ID grace period has the same length.
const UPGRADE_GRACE_PERIOD_BLOCKS: block::HeightDiff = 40;

/// Returns the network upgrade before the one active at `height`, if `height`
/// is in the first [`UPGRADE_GRACE_PERIOD_BLOCKS`] blocks of that upgrade.
///
/// Only upgrades with a consensus branch ID have a grace period, so
/// `BeforeOverwinter` does not.
fn upgrade_grace_period(network: &Network, height: block::Height) -> Option<NetworkUpgrade> {
    let (upgrade, activation_height) =
        NetworkUpgrade::current_with_activation_height(network, height);

    if upgrade.branch_id().is_none() || height - activation_height >= UPGRADE_GRACE_PERIOD_BLOCKS {
        return None;
    }

    // Several upgrades can activate at one height, so the previous upgrade is
    // the one active at the block before.
    let previous_height = activation_height.previous().ok()?;
    Some(NetworkUpgrade::current(network, previous_height))
}

/// Returns the peer to put in a transaction cooldown for `error`, if any.
///
/// [`peer_action`] decides which failures start a cooldown. Failures without a
/// legacy advertiser address do not, and neither do failures verified against
/// a tip other than `best_tip_height`. `tip_upgrade` is the network upgrade of
/// the block after `best_tip_height`. `grace_period_upgrade` is the upgrade
/// before `tip_upgrade`, if that block is in the upgrade's grace period.
fn transaction_cooldown_peer(
    error: &TransactionDownloadVerifyError,
    best_tip_height: Option<block::Height>,
    tip_upgrade: NetworkUpgrade,
    grace_period_upgrade: Option<NetworkUpgrade>,
) -> Option<PeerSocketAddr> {
    let TransactionDownloadVerifyError::Invalid {
        error,
        advertiser_addr: Some(advertiser_addr),
        tip_height,
        transaction_version,
        transaction_network_upgrade,
        retried_after_tip_reset,
    } = error
    else {
        return None;
    };

    // A reset can activate an upgrade, so a retried transaction can fail
    // where it was valid when the peer relayed it.
    if *retried_after_tip_reset {
        return None;
    }

    let cools_down = match peer_action(error) {
        PeerAction::Cooldown => true,
        PeerAction::BanV5AndLater => *transaction_version < 5,
        PeerAction::Ban | PeerAction::Ignore => false,
    };

    if !cools_down || tip_height.is_none() || *tip_height != best_tip_height {
        return None;
    }

    // Tip timestamps can be up to two hours in the future, so this node can
    // pass the tip distance check while its tip is before an upgrade that the
    // network has activated. A transaction that declares a later upgrade than
    // the tip's shows that this node can be behind, so it starts no cooldown.
    // Below NU5, the verifier does not check branch IDs, so such a transaction
    // can fail with other errors.
    if transaction_network_upgrade.is_some_and(|upgrade| upgrade > tip_upgrade) {
        return None;
    }

    // A peer whose tip lags an upgrade activation relays transactions built
    // for the previous upgrade. Those transactions can fail with signature or
    // pool rule errors. So shortly after an activation,
    // a transaction that declares the previous upgrade, or declares no
    // upgrade, starts no cooldown.
    if grace_period_upgrade.is_some_and(|previous_upgrade| {
        transaction_network_upgrade.is_none_or(|upgrade| upgrade == previous_upgrade)
    }) {
        return None;
    }

    Some(*advertiser_addr)
}

/// What the mempool does to the peer that relayed a transaction that failed
/// verification.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum PeerAction {
    /// Ban the peer. The failure only depends on the transaction and the
    /// outputs it spends.
    Ban,

    /// Ban the peer for a v5 or later transaction, and start a cooldown for a
    /// v4 transaction.
    ///
    /// The failure depends on the network upgrade that this node verifies the
    /// transaction under. A v4 transaction does not encode a consensus branch
    /// ID. A node whose tip lags the network at an upgrade activation verifies
    /// it under the wrong upgrade, so these failures can reject a valid v4
    /// transaction.
    ///
    /// A v5 or later transaction encodes its branch ID. The verifier rejects a
    /// branch ID that differs from this node's upgrade, or a transaction
    /// version that the upgrade does not support, before it runs these checks.
    /// So these failures only reach a v5 or later transaction that this node
    /// verifies under the upgrade the transaction declares.
    BanV5AndLater,

    /// Start a transaction cooldown. The failure depends on this node's tip,
    /// which can differ from the relaying peer's tip.
    Cooldown,

    /// Do nothing. The failure does not show that the relaying peer
    /// misbehaved.
    Ignore,
}

/// Returns what the mempool does to the peer that relayed a transaction that
/// failed with `error`.
///
/// The match lists every error, so each new verification failure needs a
/// decision.
fn peer_action(error: &TransactionError) -> PeerAction {
    use PeerAction::*;
    use TransactionError::*;

    match error {
        // The transaction's structure, encodings, Sprout proofs, proof sizes,
        // and value balance. An outpoint fixes the value of the output it
        // spends, so the fee does not depend on the tip either.
        WrongVersion
        | NoInputs
        | NoOutputs
        | BadBalance
        | IncorrectFee
        | SmallOrder
        | Groth16(_)
        | MalformedGroth16(_)
        | BothVPubsNonZero
        | NotEnoughFlags
        | NotEnoughIronwoodFlags
        | OrchardHasEnableCrossAddress
        | OrchardProofSize
        | IronwoodProofSize
        | CoinbaseInMempool
        | NonCoinbaseHasCoinbaseInput => Ban,

        // Spends that repeat inside one transaction, and an expiry height above
        // the maximum. Spends that conflict with the chain or the mempool fail
        // with other errors. zcashd scores these failures 100.
        DuplicateTransparentSpend(_)
        | DuplicateSproutNullifier(_)
        | DuplicateSaplingNullifier(_)
        | DuplicateOrchardNullifier(_)
        | DuplicateIronwoodNullifier(_)
        | MaximumExpiryHeight { .. } => Ban,

        // A transaction that spends a transparent coinbase output must only
        // have shielded outputs. The outpoint fixes whether the spent output
        // is a coinbase output, so the rule does not depend on the tip.
        // zcashd does not score this failure.
        UnshieldedTransparentCoinbaseSpend { .. } => Ban,
        ValidateContextError(error) => match **error {
            zs::ValidateContextError::UnshieldedTransparentCoinbaseSpend { .. } => Ban,

            // Spends and anchors in the chain, and coinbase maturity, depend
            // on the tip. zcashd does not score spent inputs or unknown
            // anchors, so that an attacker cannot make honest nodes
            // disconnect each other.
            _ => Ignore,
        },

        // Signature hashes commit to the verifying upgrade's branch ID, and
        // Halo2 proofs use that upgrade's verifying key. The Sapling verifier
        // checks proofs and signatures together. Only v4 transactions have
        // Sprout signatures, so Ed25519 failures never ban.
        Script(script_error) if script_error.is_script_failure() => BanV5AndLater,
        SaplingVerificationFailed
        | Halo2VerificationFailed
        | Ed25519(_)
        | RedJubjub(_)
        | RedPallas(_) => BanV5AndLater,

        // Only v5 and later transactions contain Orchard actions.
        DisabledAddToOrchardPool => BanV5AndLater,

        // ZIP 218 limits apply from NU7. A v5 or later transaction declares
        // its upgrade, and a branch ID mismatch fails first. A v4 transaction
        // declares none, so a peer behind NU7 can relay one that exceeds a
        // limit.
        OrchardActionsExceedBlockLimit { .. }
        | IronwoodActionsExceedBlockLimit { .. }
        | SaplingIOsExceedBlockLimit { .. }
        | SproutJoinSplitsExceedBlockLimit { .. }
        | ShieldedCostExceedsBlockBudget { .. } => BanV5AndLater,

        // Canopy forbids Sprout deposits. Replace the previous ban with a
        // cooldown because an honest peer can still be before activation.
        DisabledAddToSproutPool => Cooldown,

        // These failures already received no mempool penalty. Preserve that
        // policy when peers disagree about the tip or network upgrade.
        WrongConsensusBranchId | UnsupportedByNetworkUpgrade(..) | ExpiredTransaction { .. } => {
            Ignore
        }

        // Honest peers relay these transactions near the tip. zcashd does not
        // score non-final transactions. A branch ID that this node does not
        // know can belong to an upgrade that this node's software predates.
        WrongConsensusBranchIdNu6_3GracePeriod
        | MissingConsensusBranchId
        | LockedUntilAfterBlockHeight(_)
        | LockedUntilAfterBlockTime(_)
        | ImmatureTransparentCoinbaseSpend { .. } => Ignore,

        // Parents can arrive late or never, and this node's own state lookup
        // can fail.
        TransparentInputNotFound | ValidateMempoolLockTimeError(_) => Ignore,

        // Policy rejections: blocks can contain these transactions.
        Zip317(_)
        | NonStandardScriptSigSize { .. }
        | NonStandardScriptSigNotPushOnly { .. }
        | NonStandardInputs => Ignore,

        // Only block verification returns these. The mempool rejects coinbase
        // transactions before their coinbase checks.
        CoinbasePosition
        | CoinbaseAfterFirst
        | CoinbaseHasJoinSplit
        | CoinbaseHasSpend
        | CoinbaseHasOutputPreHeartwood
        | CoinbaseHasEnableSpendsOrchard
        | CoinbaseHasEnableSpendsIronwood
        | CoinbaseHasOrchardShieldedData
        | CoinbaseOutputsNotDecryptable
        | NotCoinbase
        | CoinbaseExpiryBlockHeight { .. }
        | CoinbaseConstruction(_)
        | Subsidy(_) => Ignore,

        // The verifier failed, or the error does not name the rule that failed.
        Script(_)
        | InternalDowncastError(_)
        | Io(_)
        | TryFromSlice(_)
        | Amount(_)
        | Balance(_)
        | Other(_) => Ignore,
    }
}

/// Returns true if `peer` is a zcashd-compat sidecar in `zcashd_compat_peer_ips`.
///
/// A sidecar syncs from this node, so its tip trails this node's tip. It
/// relays transactions that its wallet built for that older tip.
fn is_zcashd_compat_peer(zcashd_compat_peer_ips: &HashSet<IpAddr>, peer: PeerSocketAddr) -> bool {
    zcashd_compat_peer_ips.contains(&peer.ip().to_canonical())
}

/// The longest time the mempool waits for the peer set to accept a disconnect.
const DISCONNECT_TIMEOUT: Duration = Duration::from_secs(5);

/// Starts a transaction cooldown for `peer`.
///
/// Also disconnects `peer`, without banning it, when this is not its first
/// cooldown. The first cooldown can come from transactions that were already
/// queued when the first failure arrived. A later cooldown only starts after an
/// earlier one ends, so the peer kept relaying invalid transactions.
fn start_peer_cooldown(
    peer_cooldowns: &peer_cooldown::PeerCooldowns,
    outbound: &Outbound,
    peer: PeerSocketAddr,
    tx_id: UnminedTxId,
    expose_peer_addresses: bool,
) {
    let Some(cooldown) = peer_cooldowns.record_invalid_transaction(peer.ip(), Instant::now())
    else {
        return;
    };

    tracing::debug!(
        ?tx_id,
        peer = %legacy_peer_log_label(peer, expose_peer_addresses),
        ?cooldown,
        "ignoring peer transaction advertisements after an invalid transaction"
    );
    metrics::counter!("mempool.peer_cooldown.started.total").increment(1);

    if cooldown > peer_cooldown::BASE_COOLDOWN {
        // Boxing works around a compiler limitation: without it, `tokio::spawn`
        // cannot prove that the buffer's error conversion holds for every lifetime.
        let disconnect = outbound
            .clone()
            .oneshot(zn::Request::DisconnectPeer(peer))
            .boxed();
        spawn_erased(async move {
            let result = tokio::time::timeout(DISCONNECT_TIMEOUT, disconnect).await;
            if !matches!(result, Ok(Ok(_))) {
                tracing::debug!("could not disconnect a peer after repeated transaction failures");
            }
        });
    }
}

/// The state of the mempool.
///
/// Indicates whether it is enabled or disabled and, if enabled, contains
/// the necessary data to run it.
//
// Zakura only has one mempool, so the enum variant size difference doesn't matter.
#[allow(clippy::large_enum_variant)]
#[derive(Default)]
enum ActiveState {
    /// The Mempool is disabled.
    #[default]
    Disabled,

    /// The Mempool is enabled.
    Enabled {
        /// The Mempool storage itself.
        ///
        /// # Correctness
        ///
        /// Only components internal to the [`Mempool`] struct are allowed to
        /// inject transactions into `storage`, as transactions must be verified beforehand.
        storage: Storage,

        /// The transaction download and verify stream.
        tx_downloads: Pin<Box<InboundTxDownloads>>,

        /// Verified transaction IDs awaiting proactive advertisement through
        /// the peer set.
        ///
        /// This pending set is separate from the full mempool inventory served
        /// by [`Request::TransactionIds`] when a peer requests it.
        pending_gossip_tx_ids: HashSet<UnminedTxId>,

        /// Last seen chain tip hash that mempool transactions have been verified against.
        ///
        /// In some tests, this is initialized to the latest chain tip, then updated in `poll_ready()` before each request.
        last_seen_tip_hash: block::Hash,
    },
}

impl ActiveState {
    /// Returns the current state, leaving [`Self::Disabled`] in its place.
    fn take(&mut self) -> Self {
        std::mem::take(self)
    }

    /// Returns a list of requests that will retry every stored and pending transaction.
    fn transaction_retry_requests(&self) -> Vec<(Gossip, Option<QueueSource>)> {
        match self {
            ActiveState::Disabled => Vec::new(),
            ActiveState::Enabled {
                storage,
                tx_downloads,
                ..
            } => {
                let mut transactions = Vec::new();

                let storage = storage
                    .transactions()
                    .values()
                    .map(|tx| (tx.transaction.clone().into(), None));
                transactions.extend(storage);

                let pending = tx_downloads
                    .transaction_requests()
                    .map(|(tx, source)| (tx.clone(), source.clone()));
                transactions.extend(pending);

                transactions
            }
        }
    }
}

/// Mempool async management and query service.
///
/// The mempool is the set of all verified transactions that this node is aware
/// of that have yet to be confirmed by the Zcash network. A transaction is
/// confirmed when it has been included in a block ('mined').
pub struct Mempool {
    /// The configurable options for the mempool, persisted between states.
    config: Config,

    /// Whether legacy peer address labels in logs are unredacted.
    expose_peer_addresses: bool,

    /// The state of the mempool.
    active_state: ActiveState,

    /// Allows checking if we are near the tip to enable/disable the mempool.
    sync_status: SyncStatus,

    /// If the state's best chain tip has reached this height, always enable the mempool.
    debug_enable_at_height: Option<Height>,

    /// Allows efficient access to the best tip of the blockchain.
    latest_chain_tip: zs::LatestChainTip,

    /// Allows the detection of newly added chain tip blocks,
    /// and chain tip resets.
    chain_tip_change: ChainTipChange,

    /// Handle to the outbound service.
    /// Used to construct the transaction downloader.
    outbound: Outbound,

    /// Handle to the read state service.
    /// Used to construct the transaction downloader.
    read_state: ReadState,

    /// Keeps the read-write state service alive for the mempool lifetime.
    /// The transaction downloader does not send requests through this handle.
    _state_guard: State,

    /// Handle to the transaction verifier service.
    /// Used to construct the transaction downloader.
    tx_verifier: TxVerifier,

    /// Sender part of a gossip transactions channel.
    /// Used to broadcast transaction ids to peers.
    transaction_sender: broadcast::Sender<MempoolChange>,

    /// Peers whose transactions are ignored after they relayed invalid
    /// transactions. Shared with the download tasks, and kept across mempool
    /// resets and deactivations.
    peer_cooldowns: peer_cooldown::PeerCooldowns,

    /// Bans peers that relay transactions whose verification failures do not
    /// depend on this node's tip.
    misbehavior_sender: mpsc::Sender<(PeerSocketAddr, u32)>,

    /// The canonical IPs of zcashd-compat sidecars. The mempool never bans,
    /// cools down, or disconnects these peers.
    zcashd_compat_peer_ips: HashSet<IpAddr>,
}

impl Mempool {
    #[allow(clippy::too_many_arguments)]
    pub(crate) fn new(
        config: &Config,
        expose_peer_addresses: bool,
        outbound: Outbound,
        state: State,
        read_state: ReadState,
        tx_verifier: TxVerifier,
        sync_status: SyncStatus,
        latest_chain_tip: zs::LatestChainTip,
        chain_tip_change: ChainTipChange,
        misbehavior_sender: mpsc::Sender<(PeerSocketAddr, u32)>,
        zcashd_compat_peer_ips: Vec<IpAddr>,
    ) -> (Self, MempoolTxSubscriber) {
        let (transaction_sender, _) =
            tokio::sync::broadcast::channel(gossip::MEMPOOL_CHANGE_CHANNEL_CAPACITY);
        let transaction_subscriber = MempoolTxSubscriber::new(transaction_sender.clone());

        let mut service = Mempool {
            config: config.clone(),
            expose_peer_addresses,
            active_state: ActiveState::Disabled,
            sync_status,
            debug_enable_at_height: config.debug_enable_at_height.map(Height),
            latest_chain_tip,
            chain_tip_change,
            outbound,
            read_state,
            _state_guard: state,
            tx_verifier,
            transaction_sender,
            peer_cooldowns: peer_cooldown::PeerCooldowns::default(),
            misbehavior_sender,
            zcashd_compat_peer_ips: zcashd_compat_peer_ips
                .into_iter()
                .map(|ip| ip.to_canonical())
                .collect(),
        };

        // Make sure `is_enabled` is accurate.
        // Otherwise, it is only updated in `poll_ready`, right before each service call.
        service.update_state(None, service.is_caught_up_to_start());

        (service, transaction_subscriber)
    }

    /// Is the mempool enabled by a debug config option?
    fn is_enabled_by_debug(&self) -> bool {
        let mut is_debug_enabled = false;

        // optimise non-debug performance
        if self.debug_enable_at_height.is_none() {
            return is_debug_enabled;
        }

        let enable_at_height = self
            .debug_enable_at_height
            .expect("unexpected debug_enable_at_height: just checked for None");

        if let Some(best_tip_height) = self.latest_chain_tip.best_tip_height() {
            is_debug_enabled = best_tip_height >= enable_at_height;

            if is_debug_enabled && !self.is_enabled() {
                info!(
                    ?best_tip_height,
                    ?enable_at_height,
                    "enabling mempool for debugging"
                );
            }
        }

        is_debug_enabled
    }

    /// Returns `true` if Zakura is caught up enough to start the mempool.
    ///
    /// Mempool activation needs both sync throughput to have slowed down and an
    /// independent local-clock estimate that the state tip is close to the
    /// network tip. The estimate prevents a peer-starved syncer from looking
    /// caught up just because it is downloading zero blocks per round.
    fn is_caught_up_to_start(&self) -> bool {
        self.is_enabled_by_debug() || self.is_current_enough_for_mempool()
    }

    /// Returns true when sync throughput and the tip estimate permit mempool use.
    fn is_current_enough_for_mempool(&self) -> bool {
        self.sync_status.is_close_to_tip()
            && is_estimated_close_to_network_tip(&self.chain_tip_change)
    }

    /// Returns the estimated distance to the network tip, if a state tip exists.
    fn estimated_distance_to_network_tip(&self) -> Option<block::HeightDiff> {
        self.chain_tip_change
            .estimate_distance_to_network_chain_tip()
            .map(|(distance, _height)| distance)
    }

    /// Returns true when the node has fallen far enough behind to disable an
    /// already-active mempool.
    fn is_far_enough_to_disable(&self) -> bool {
        !self.is_enabled_by_debug()
            && !self.chain_tip_change.network().is_a_test_network()
            && self
                .estimated_distance_to_network_tip()
                .is_none_or(|distance| distance > i64::from(zs::MAX_BLOCK_REORG_HEIGHT))
    }

    /// Replaces the active state with a freshly-initialised [`ActiveState::Enabled`],
    /// using `tip_action`'s best tip hash as the `last_seen_tip_hash`.
    fn enable_at_tip(&mut self, tip_action: &TipAction, reason: &'static str) {
        let (last_seen_tip_hash, tip_height) = tip_action.best_tip_hash_and_height();

        info!(?tip_height, reason, "activating mempool");

        let tx_downloads = Box::pin(TxDownloads::new(
            Timeout::new(self.outbound.clone(), TRANSACTION_DOWNLOAD_TIMEOUT),
            Timeout::new(self.tx_verifier.clone(), TRANSACTION_VERIFY_TIMEOUT),
            self.read_state.clone(),
            self.expose_peer_addresses,
            self.config.max_transaction_bytes,
            self.peer_cooldowns.clone(),
        ));
        self.active_state = ActiveState::Enabled {
            storage: storage::Storage::new(&self.config),
            tx_downloads,
            pending_gossip_tx_ids: HashSet::new(),
            last_seen_tip_hash,
        };
    }

    /// Update the mempool state (enabled / disabled) depending on how close to
    /// the tip is the synchronization, including side effects to state changes.
    ///
    /// Accepts an optional [`TipAction`] for setting the `last_seen_tip_hash` field
    /// when enabling the mempool state, it will not enable the mempool if this is None.
    /// Uses the activation decision sampled before consuming the tip action.
    ///
    /// Returns `true` if the state changed.
    fn update_state(
        &mut self,
        tip_action: Option<&TipAction>,
        is_caught_up_to_start: bool,
    ) -> bool {
        let is_far_enough_to_disable = self.is_far_enough_to_disable();

        match (
            is_caught_up_to_start,
            is_far_enough_to_disable,
            self.is_enabled(),
            tip_action,
        ) {
            // the active state is up to date, or there is no tip action to activate the mempool
            (false, _, false, _) | (true, _, true, _) | (true, _, false, None) => return false,

            // Enable state - there should be a chain tip when Zakura is close
            // to the network tip.
            (true, _, false, Some(tip_action)) => self.enable_at_tip(
                tip_action,
                if self.is_enabled_by_debug() {
                    "debug height reached"
                } else {
                    "estimated close to the network tip"
                },
            ),

            // Disable once the same clock estimate used for activation falls
            // outside the rollback window.
            (_, true, true, _) => {
                let estimated_distance_to_network_tip = self.estimated_distance_to_network_tip();
                info!(
                    ?estimated_distance_to_network_tip,
                    disable_distance = zs::MAX_BLOCK_REORG_HEIGHT,
                    "deactivating mempool: Zakura is far from the network tip"
                );

                if let ActiveState::Enabled { storage, .. } = &self.active_state {
                    let invalidated_ids: HashSet<_> = storage.tx_ids().collect();
                    if !invalidated_ids.is_empty() {
                        let _ = self
                            .transaction_sender
                            .send(MempoolChange::invalidated(invalidated_ids));
                    }
                }
                self.active_state = ActiveState::Disabled;
            }

            (false, false, true, _) => return false,
        };

        true
    }

    /// Return whether the mempool is enabled or not.
    pub fn is_enabled(&self) -> bool {
        match self.active_state {
            ActiveState::Disabled => false,
            ActiveState::Enabled { .. } => true,
        }
    }
}

impl Service<Request> for Mempool {
    type Response = Response;
    type Error = BoxError;
    type Future =
        Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'static>>;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        let is_caught_up_to_start = self.is_caught_up_to_start();
        let should_check_tip = self.is_enabled() || is_caught_up_to_start;
        let tip_action = should_check_tip
            .then(|| self.chain_tip_change.last_tip_change())
            .flatten();

        let is_state_changed = self.update_state(tip_action.as_ref(), is_caught_up_to_start);

        tracing::trace!(is_enabled = ?self.is_enabled(), ?is_state_changed, "started polling the mempool...");

        // When the mempool is disabled we still return that the service is ready.
        // Otherwise, callers could block waiting for the mempool to be enabled.
        if !self.is_enabled() {
            return Poll::Ready(Ok(()));
        }

        // Clear the mempool and cancel downloads if there has been a chain tip reset.
        //
        // But if the mempool was just freshly enabled,
        // skip resetting and removing mined transactions for this tip.
        if !is_state_changed && matches!(tip_action, Some(TipAction::Reset { .. })) {
            info!(
                tip_height = ?tip_action.as_ref().unwrap().best_tip_height(),
                "resetting mempool: switched best chain, skipped blocks, or activated network upgrade"
            );

            let previous_state = self.active_state.take();
            let tx_retries = previous_state.transaction_retry_requests();

            // Use the same code for dropping and resetting the mempool,
            // to avoid subtle bugs.
            //
            // Drop the current contents of the state,
            // cancelling any pending download tasks,
            // and dropping completed verification results.
            std::mem::drop(previous_state);

            // Re-initialise an empty state. `update_state()` has already applied
            // the distance-based disable gate, so an active mempool can safely
            // reset at the current tip.
            self.enable_at_tip(
                tip_action
                    .as_ref()
                    .expect("this branch only matches when tip_action is a Reset"),
                "chain tip reset",
            );

            // Re-verify the transactions that were pending or valid at the previous tip.
            // This saves us the time and data needed to re-download them.
            if let ActiveState::Enabled { tx_downloads, .. } = &mut self.active_state {
                info!(
                    transactions = tx_retries.len(),
                    "re-verifying mempool transactions after a chain fork"
                );

                for (tx, source) in tx_retries {
                    // This is just an efficiency optimisation, so we don't care if queueing
                    // transaction requests fails.
                    let _result = tx_downloads.retry_after_tip_reset(tx, source);
                }
            }

            return Poll::Ready(Ok(()));
        }

        let is_current_enough_for_mempool = self.is_current_enough_for_mempool();

        if let ActiveState::Enabled {
            storage,
            tx_downloads,
            pending_gossip_tx_ids,
            last_seen_tip_hash,
        } = &mut self.active_state
        {
            // Collect inserted transaction ids.
            let mut send_to_peers_ids = HashSet::<_>::new();
            let mut invalidated_ids = HashSet::<_>::new();
            let mut mined_mempool_ids = HashSet::<_>::new();

            let best_tip_height = self.latest_chain_tip.best_tip_height();
            let next_height = best_tip_height
                .and_then(|height| height + 1)
                .unwrap_or(Height(0));
            let network = self.chain_tip_change.network();
            let tip_upgrade = NetworkUpgrade::current(network, next_height);
            let grace_period_upgrade = upgrade_grace_period(network, next_height);

            // Clean up completed download tasks and add to mempool if successful.
            while let Poll::Ready(Some(result)) = pin!(&mut *tx_downloads).poll_next(cx) {
                match result {
                    Ok(Ok((tx, spent_mempool_outpoints, expected_tip_height, rsp_tx))) => {
                        // # Correctness:
                        //
                        // It's okay to use tip height here instead of the tip hash since
                        // chain_tip_change.last_tip_change() returns a `TipAction::Reset` when
                        // the best chain changes (which is the only way to stay at the same height), and the
                        // mempool re-verifies all pending tx_downloads when there's a `TipAction::Reset`.
                        if best_tip_height == expected_tip_height {
                            let tx_id = tx.transaction.id();
                            let (insert_result, evicted_ids) = storage.insert_with_evicted_ids(
                                tx,
                                spent_mempool_outpoints,
                                best_tip_height,
                            );

                            tracing::trace!(
                                ?insert_result,
                                ?evicted_ids,
                                "got Ok(_) transaction verify, tried to store",
                            );

                            if let Ok(inserted_id) = insert_result {
                                // Save transaction ids that we will send to peers
                                send_to_peers_ids.insert(inserted_id);
                            } else {
                                invalidated_ids.insert(tx_id);
                            }

                            if !evicted_ids.is_empty() {
                                // A later insertion can evict a transaction accepted earlier in
                                // this `poll_ready` pass, so do not advertise it.
                                send_to_peers_ids.retain(|id| !evicted_ids.contains(id));
                                invalidated_ids.extend(evicted_ids);
                            }

                            // Send the result to responder channel if one was provided.
                            if let Some(rsp_tx) = rsp_tx {
                                let _ = rsp_tx
                                    .send(insert_result.map(|_| ()).map_err(|err| err.into()));
                            }
                        } else {
                            tracing::trace!("chain grew during tx verification, retrying ..",);

                            // We don't care if re-queueing the transaction request fails.
                            let _result = tx_downloads.download_if_needed_and_verify(
                                tx.transaction.into(),
                                None,
                                rsp_tx,
                            );
                        }
                    }
                    Ok(Err(boxed_err)) => {
                        let (tx_id, error) = *boxed_err;
                        let zcashd_compat_peer_ips = &self.zcashd_compat_peer_ips;
                        let ban_peer = transaction_ban_peer(&error)
                            .filter(|peer| !is_zcashd_compat_peer(zcashd_compat_peer_ips, *peer));
                        if let Some(peer) = ban_peer {
                            let _ = self
                                .misbehavior_sender
                                .try_send((peer, zn::constants::MAX_PEER_MISBEHAVIOR_SCORE));
                        }

                        // Only start cooldowns while this node's validation context is
                        // current. A stale node verifies transactions against old rules.
                        let cooldown_peer = is_current_enough_for_mempool
                            .then(|| {
                                transaction_cooldown_peer(
                                    &error,
                                    best_tip_height,
                                    tip_upgrade,
                                    grace_period_upgrade,
                                )
                            })
                            .flatten()
                            .filter(|peer| !is_zcashd_compat_peer(zcashd_compat_peer_ips, *peer));
                        if let Some(peer) = cooldown_peer {
                            start_peer_cooldown(
                                &self.peer_cooldowns,
                                &self.outbound,
                                peer,
                                tx_id,
                                self.expose_peer_addresses,
                            );
                        }

                        let peer_label =
                            transaction_error_peer_log_label(&error, self.expose_peer_addresses);
                        let peer = peer_label.as_deref().unwrap_or("none");
                        tracing::debug!(
                            ?tx_id,
                            ?error,
                            peer = %peer,
                            "mempool transaction was not accepted"
                        );

                        match &error {
                            TransactionDownloadVerifyError::PolicyRejected(
                                storage::NonStandardTransactionError::TransactionTooLarge {
                                    ..
                                },
                            ) => metrics::counter!(
                                "mempool.rejected.transactions.total",
                                "reason" => "transaction_too_large"
                            )
                            .increment(1),
                            TransactionDownloadVerifyError::PeerCoolingDown => metrics::counter!(
                                "mempool.peer_cooldown.ignored.transactions.total"
                            )
                            .increment(1),
                            _ => metrics::counter!(
                                "mempool.failed.verify.tasks.total",
                                "reason" => error.to_string()
                            )
                            .increment(1),
                        }

                        invalidated_ids.insert(tx_id);
                        storage.reject_if_needed(tx_id, error);
                    }
                    Err((tx_id, _elapsed)) => {
                        tracing::info!(
                            ?tx_id,
                            "mempool transaction failed to verify due to timeout"
                        );

                        invalidated_ids.insert(tx_id);

                        metrics::counter!("mempool.failed.verify.tasks.total", "reason" => "timeout").increment(1);
                    }
                };
            }

            // Handle best chain tip changes
            if let Some(TipAction::Grow { block }) = tip_action {
                tracing::trace!(block_height = ?block.height, "handling blocks added to tip");
                *last_seen_tip_hash = block.hash;

                // Cancel downloads/verifications/storage of transactions
                // with the same mined IDs as recently mined transactions.
                let mined_ids = block.transaction_hashes.iter().cloned().collect();
                tx_downloads.cancel(&mined_ids);
                storage.clear_mined_dependencies(&mined_ids);

                let storage::RemovedTransactionIds { mined, invalidated } =
                    storage.reject_and_remove_same_effects(&mined_ids, block.transactions);

                // Clear any transaction rejections if they might have become valid after
                // the new block was added to the tip.
                storage.clear_tip_rejections();

                mined_mempool_ids.extend(mined);
                invalidated_ids.extend(invalidated);
            }

            // Remove expired transactions from the mempool.
            //
            // Lock times never expire, because block times are strictly increasing.
            // So we don't need to check them here.
            if let Some(tip_height) = best_tip_height {
                let expired_transactions = storage.remove_expired_transactions(tip_height);
                // Remove transactions that are expired from the peers list
                send_to_peers_ids.retain(|id| !expired_transactions.contains(id));

                if !expired_transactions.is_empty() {
                    tracing::debug!(
                        ?expired_transactions,
                        "removed expired transactions from the mempool",
                    );

                    invalidated_ids.extend(expired_transactions);
                }
            }

            // Send transactions that were not rejected nor expired to peers and RPC listeners.
            if !send_to_peers_ids.is_empty() {
                tracing::trace!(
                    ?send_to_peers_ids,
                    "sending new transactions to peers and RPC listeners"
                );

                pending_gossip_tx_ids.extend(send_to_peers_ids.iter().copied());

                self.transaction_sender
                    .send(MempoolChange::added(send_to_peers_ids))?;
            }

            // Prune transaction IDs that no longer need gossip from the pending set.
            //
            // This runs after the pending set is extended, so it also covers
            // transactions that were inserted and then invalidated or mined
            // within this same `poll_ready` call.
            if !invalidated_ids.is_empty() || !mined_mempool_ids.is_empty() {
                pending_gossip_tx_ids.retain(|tx_id| {
                    !invalidated_ids.contains(tx_id) && !mined_mempool_ids.contains(tx_id)
                });
            }

            // Send transactions that were rejected to RPC listeners.
            if !invalidated_ids.is_empty() {
                tracing::trace!(
                    ?invalidated_ids,
                    "sending invalidated transactions to RPC listeners"
                );

                self.transaction_sender
                    .send(MempoolChange::invalidated(invalidated_ids))?;
            }

            // Send transactions that were mined onto the best chain to RPC listeners.
            if !mined_mempool_ids.is_empty() {
                tracing::trace!(
                    ?mined_mempool_ids,
                    "sending mined transactions to RPC listeners"
                );

                self.transaction_sender
                    .send(MempoolChange::mined(mined_mempool_ids))?;
            }
        }

        Poll::Ready(Ok(()))
    }

    /// Call the mempool service.
    ///
    /// Errors indicate that the peer has done something wrong or unexpected,
    /// and will cause callers to disconnect from the remote peer.
    #[instrument(name = "mempool", skip(self, req))]
    fn call(&mut self, req: Request) -> Self::Future {
        match &mut self.active_state {
            ActiveState::Enabled {
                storage,
                tx_downloads,
                pending_gossip_tx_ids,
                last_seen_tip_hash,
            } => match req {
                // Queries
                Request::TransactionIds => {
                    trace!(?req, "got mempool request");

                    let res: HashSet<_> = storage.tx_ids().collect();

                    trace!(?req, res_count = ?res.len(), "answered mempool request");

                    async move { Ok(Response::TransactionIds(res)) }.boxed()
                }

                Request::TakePendingGossipTransactionIds { limit } => {
                    trace!(?req, "got mempool request");

                    let res = if pending_gossip_tx_ids.len() <= limit {
                        std::mem::take(pending_gossip_tx_ids)
                    } else {
                        let res: HashSet<_> =
                            pending_gossip_tx_ids.iter().copied().take(limit).collect();

                        for tx_id in &res {
                            pending_gossip_tx_ids.remove(tx_id);
                        }

                        res
                    };

                    trace!(?req, res_count = ?res.len(), "answered mempool request");

                    async move { Ok(Response::TransactionIds(res)) }.boxed()
                }

                Request::TransactionsById(ref ids) => {
                    trace!(?req, "got mempool request");

                    let res: Vec<_> = storage.transactions_exact(ids.clone()).cloned().collect();

                    trace!(?req, res_count = ?res.len(), "answered mempool request");

                    async move { Ok(Response::Transactions(res)) }.boxed()
                }
                Request::TransactionsByMinedId(ref ids) => {
                    trace!(?req, "got mempool request");

                    let res: Vec<_> = storage
                        .transactions_same_effects(ids.clone())
                        .cloned()
                        .collect();

                    trace!(?req, res_count = ?res.len(), "answered mempool request");

                    async move { Ok(Response::Transactions(res)) }.boxed()
                }
                Request::TransactionWithDepsByMinedId(tx_id) => {
                    trace!(?req, "got mempool request");

                    let res = if let Some((transaction, dependencies)) =
                        storage.transaction_with_deps(tx_id)
                    {
                        Ok(Response::TransactionWithDeps {
                            transaction,
                            dependencies,
                        })
                    } else {
                        Err("transaction not found in mempool".into())
                    };

                    trace!(?req, ?res, "answered mempool request");

                    async move { res }.boxed()
                }

                Request::AwaitOutput(outpoint) => {
                    trace!(?req, "got mempool request");

                    let response_fut = storage.pending_outputs.queue(outpoint);

                    if let Some(output) = storage.created_output(&outpoint) {
                        storage.pending_outputs.respond(&outpoint, output)
                    }

                    trace!("answered mempool request");

                    response_fut.boxed()
                }

                Request::FullTransactions => {
                    trace!(?req, "got mempool request");

                    let transactions: Vec<_> = storage.transactions().values().cloned().collect();
                    let transaction_dependencies = storage.transaction_dependencies().clone();

                    trace!(?req, transactions_count = ?transactions.len(), "answered mempool request");

                    let response = Response::FullTransactions {
                        transactions,
                        transaction_dependencies,
                        last_seen_tip_hash: *last_seen_tip_hash,
                    };

                    async move { Ok(response) }.boxed()
                }

                Request::RejectedTransactionIds(ref ids) => {
                    trace!(?req, "got mempool request");

                    let res = storage.rejected_transactions(ids.clone()).collect();

                    trace!(?req, ?res, "answered mempool request");

                    async move { Ok(Response::RejectedTransactionIds(res)) }.boxed()
                }

                // Queue mempool candidates
                Request::Queue(gossiped_txs) => {
                    trace!(req_count = ?gossiped_txs.len(), "got mempool Queue request");

                    let rsp: Vec<Result<oneshot::Receiver<Result<(), BoxError>>, BoxError>> =
                        gossiped_txs
                            .into_iter()
                            .map(
                                |gossiped_tx| -> Result<
                                    oneshot::Receiver<Result<(), BoxError>>,
                                    MempoolError,
                                > {
                                    let (rsp_tx, rsp_rx) = oneshot::channel();
                                    match storage.should_download_or_verify(gossiped_tx.id()) {
                                        Ok(()) => {}
                                        Err(
                                            error @ MempoolError::StorageExactTip(
                                                ExactTipRejectionError::FailedStandard(
                                                    storage::NonStandardTransactionError::TransactionTooLarge {
                                                        ..
                                                    },
                                                ),
                                            ),
                                        ) => {
                                            // A cached policy rejection is a completed admission
                                            // result, not a failure to queue the request.
                                            let _ = rsp_tx.send(Err(error.into()));
                                            return Ok(rsp_rx);
                                        }
                                        Err(error) => return Err(error),
                                    }
                                    tx_downloads.download_if_needed_and_verify(
                                        gossiped_tx,
                                        None,
                                        Some(rsp_tx),
                                    )?;

                                    Ok(rsp_rx)
                                },
                            )
                            .map(|result| result.map_err(BoxError::from))
                            .collect();

                    async move { Ok(Response::Queued(rsp)) }.boxed()
                }

                // Queue transaction candidates received from a specific peer,
                // whether inv-advertised ids or directly pushed transactions.
                // Per-peer accounting is enforced inside the downloader.
                Request::QueueFromPeer {
                    transactions,
                    source,
                } => {
                    trace!(
                        req_count = ?transactions.len(),
                        source = %queue_source_log_label(&source, self.expose_peer_addresses),
                        "got mempool QueueFromPeer request"
                    );

                    let is_cooling_down = match &source {
                        QueueSource::LegacySocket(addr) => self
                            .peer_cooldowns
                            .is_cooling_down(addr.ip(), Instant::now()),
                        QueueSource::Zakura(_) => false,
                    };

                    if is_cooling_down {
                        trace!(
                            req_count = ?transactions.len(),
                            "ignored transactions from a peer in a transaction cooldown"
                        );
                        metrics::counter!("mempool.peer_cooldown.ignored.transactions.total")
                            .increment(u64::try_from(transactions.len()).unwrap_or(u64::MAX));
                    } else {
                        for gossiped_tx in transactions {
                            if storage.should_download_or_verify(gossiped_tx.id()).is_err() {
                                continue;
                            }
                            let _ = tx_downloads.download_if_needed_and_verify(
                                gossiped_tx,
                                Some(source.clone()),
                                None,
                            );
                        }
                    }

                    async move { Ok(Response::Queued(Vec::new())) }.boxed()
                }

                // Store successfully downloaded and verified transactions in the mempool
                Request::CheckForVerifiedTransactions => {
                    trace!(?req, "got mempool request");

                    // all the work for this request is done in poll_ready
                    async move { Ok(Response::CheckedForVerifiedTransactions) }.boxed()
                }

                // Summary statistics for the mempool: count, total size, and memory usage.
                //
                // Used by the `getmempoolinfo` RPC method
                Request::QueueStats => {
                    trace!(?req, "got mempool request");

                    let size = storage.transaction_count();

                    let bytes = storage.total_serialized_size();

                    let usage = bytes; // TODO: Placeholder, should be fixed later

                    // TODO: Set to Some(true) on regtest once network info is available.
                    let fully_notified = None;

                    trace!(size, bytes, usage, "answered mempool request");

                    async move {
                        Ok(Response::QueueStats {
                            size,
                            bytes,
                            usage,
                            fully_notified,
                        })
                    }
                    .boxed()
                }
                Request::UnspentOutput(outpoint) => {
                    trace!(?req, "got mempool request");

                    if storage.has_spent_outpoint(&outpoint) {
                        trace!(?req, "answered mempool request");

                        return async move {
                            Ok(Response::TransparentOutput(Some(CreatedOrSpent::Spent)))
                        }
                        .boxed();
                    }

                    if let Some((tx_version, output)) = storage
                        .transactions()
                        .get(&outpoint.hash)
                        .map(|tx| tx.transaction.transaction().clone())
                        .and_then(|tx| {
                            tx.outputs()
                                .get(outpoint.index as usize)
                                .map(|output| (tx.version(), output.clone()))
                        })
                    {
                        trace!(?req, "answered mempool request");

                        let last_seen_hash = *last_seen_tip_hash;
                        return async move {
                            Ok(Response::TransparentOutput(Some(CreatedOrSpent::Created {
                                output,
                                tx_version,
                                last_seen_hash,
                            })))
                        }
                        .boxed();
                    }

                    trace!(?req, "answered mempool request");

                    async move { Ok(Response::TransparentOutput(None)) }.boxed()
                }
            },
            ActiveState::Disabled => {
                // TODO: add the name of the request, but not the content,
                //       like the command() or Display impls of network requests
                trace!("got mempool request while mempool is disabled");

                // We can't return an error since that will cause a disconnection
                // by the peer connection handler. Therefore, return successful
                // empty responses.

                let resp = match req {
                    // Return empty responses for queries.
                    Request::TransactionIds | Request::TakePendingGossipTransactionIds { .. } => {
                        Response::TransactionIds(Default::default())
                    }

                    Request::TransactionsById(_) => Response::Transactions(Default::default()),
                    Request::TransactionsByMinedId(_) => Response::Transactions(Default::default()),
                    Request::TransactionWithDepsByMinedId(_)
                    | Request::AwaitOutput(_)
                    | Request::UnspentOutput(_) => {
                        return async move { Err(MempoolDisabledError.into()) }.boxed()
                    }

                    Request::FullTransactions => {
                        return async move { Err(MempoolDisabledError.into()) }.boxed()
                    }

                    Request::RejectedTransactionIds(_) => {
                        Response::RejectedTransactionIds(Default::default())
                    }

                    // Don't queue mempool candidates, because there is no queue.
                    Request::Queue(gossiped_txs) => Response::Queued(
                        // Special case; we can signal the error inside the response,
                        // because the inbound service ignores inner errors.
                        iter::repeat_n(MempoolError::Disabled, gossiped_txs.len())
                            .map(BoxError::from)
                            .map(Err)
                            .collect(),
                    ),

                    // Drop peer-advertised txids when the mempool is disabled.
                    Request::QueueFromPeer { .. } => Response::Queued(Vec::new()),

                    // Check if the mempool should be enabled.
                    // This request makes sure mempools are debug-enabled in the acceptance tests.
                    Request::CheckForVerifiedTransactions => {
                        // all the work for this request is done in poll_ready
                        Response::CheckedForVerifiedTransactions
                    }

                    // Return empty mempool stats
                    Request::QueueStats => Response::QueueStats {
                        size: 0,
                        bytes: 0,
                        usage: 0,
                        fully_notified: None,
                    },
                };

                async move { Ok(resp) }.boxed()
            }
        }
    }
}