mempool: make the local-orphan capacity configurable

Address OCR findings on PR #2126:
- local_orphan_pool_capacity becomes a MempoolConfig setting (default
  DEFAULT_LOCAL_ORPHAN_POOL_CAPACITY = 25), exposed through the mempool
  RPC config and the node config file, like the sibling limits;
  validated to be non-zero in MempoolConfig::validate;
- OrphanPoolError::LocalCapacityExceeded carries the configured limit,
  mirroring TooLarge;
- the on_new_tip work-queue drain is bounded to 64 work units per event
  to keep per-event latency bounded on deep reorgs; the rest of the
  queue is processed by the round-based scheduler.

Author: nullPointerEnjoyer

blockprod/src/detail/tests/produce_block/tx_selection_by_deps.rs

@@ -119,6 +119,7 @@ async fn token_account_deps(#[case] seed: Seed, fees_selection: FeesSelection) {
         .with_chain_config(Arc::clone(&chain_config))
         .with_mempool_config(MempoolConfig {
             allow_local_orphans: Default::default(),
+            local_orphan_pool_capacity: Default::default(),
             min_tx_relay_fee_rate: FeeRate::from_amount_per_kb(Amount::ZERO).into(),
             max_cluster_tx_count: Default::default(),
             max_cluster_size_bytes: Default::default(),
@@ -578,6 +579,7 @@ async fn pool_creation_and_decommissioning(#[case] seed: Seed) {
         .with_chain_config(Arc::clone(&chain_config))
         .with_mempool_config(MempoolConfig {
             allow_local_orphans: Default::default(),
+            local_orphan_pool_capacity: Default::default(),
             min_tx_relay_fee_rate: FeeRate::from_amount_per_kb(Amount::ZERO).into(),
             max_cluster_tx_count: Default::default(),
             max_cluster_size_bytes: Default::default(),
@@ -709,6 +711,7 @@ async fn pool_creation_and_delegation(#[case] seed: Seed, fees_selection: FeesSe
         .with_chain_config(Arc::clone(&chain_config))
         .with_mempool_config(MempoolConfig {
             allow_local_orphans: Default::default(),
+            local_orphan_pool_capacity: Default::default(),
             min_tx_relay_fee_rate: FeeRate::from_amount_per_kb(Amount::ZERO).into(),
             max_cluster_tx_count: Default::default(),
             max_cluster_size_bytes: Default::default(),
@@ -934,6 +937,7 @@ async fn pool_creation_and_delegation_withdrawals(
         .with_chain_config(Arc::clone(&chain_config))
         .with_mempool_config(MempoolConfig {
             allow_local_orphans: Default::default(),
+            local_orphan_pool_capacity: Default::default(),
             min_tx_relay_fee_rate: FeeRate::from_amount_per_kb(Amount::ZERO).into(),
             max_cluster_tx_count: Default::default(),
             max_cluster_size_bytes: Default::default(),
@@ -1098,6 +1102,7 @@ async fn order_creation_and_usage(
         .with_chain_config(Arc::clone(&chain_config))
         .with_mempool_config(MempoolConfig {
             allow_local_orphans: Default::default(),
+            local_orphan_pool_capacity: Default::default(),
             min_tx_relay_fee_rate: FeeRate::from_amount_per_kb(Amount::ZERO).into(),
             max_cluster_tx_count: Default::default(),
             max_cluster_size_bytes: Default::default(),

mempool/src/config.rs

@@ -148,6 +148,14 @@ make_config_setting!(
 make_config_setting!(MaxClusterTxCount, usize, 64);
 make_config_setting!(MaxClusterSizeBytes, usize, 100_000);
 
+// Separate cap for local-origin orphans, so that they cannot starve remote peers'
+// orphans out of the main orphan pool when `AllowLocalOrphans` is enabled.
+make_config_setting!(
+    LocalOrphanPoolCapacity,
+    usize,
+    DEFAULT_LOCAL_ORPHAN_POOL_CAPACITY
+);
+
 // When disabled (the default), transactions submitted locally whose inputs are not yet known are
 // rejected outright. When enabled, they are parked in the orphan pool instead and re-processed
 // once their parents arrive (or are mined), same as remote transactions.
@@ -166,6 +174,9 @@ pub struct MempoolConfig {
 
     /// Whether to park local-origin transactions with unknown inputs in the orphan pool.
     pub allow_local_orphans: AllowLocalOrphans,
+
+    /// Capacity for local-origin orphans within the orphan pool.
+    pub local_orphan_pool_capacity: LocalOrphanPoolCapacity,
 }
 
 impl MempoolConfig {
@@ -179,13 +190,15 @@ impl MempoolConfig {
             max_cluster_tx_count,
             max_cluster_size_bytes,
             allow_local_orphans,
+            local_orphan_pool_capacity,
         } = self;
 
         RpcMempoolConfig {
             min_tx_relay_fee_rate: **min_tx_relay_fee_rate,
             max_cluster_tx_count: **max_cluster_tx_count,
             max_cluster_size_bytes: **max_cluster_size_bytes,
             allow_local_orphans: **allow_local_orphans,
+            local_orphan_pool_capacity: **local_orphan_pool_capacity,
         }
     }
 
@@ -202,6 +215,17 @@ impl MempoolConfig {
             *self.max_cluster_size_bytes > 0,
             ConfigError::MaxClusterSizeBytesCannotBeZero
         );
+        // Note: a value of 0 combined with `allow_local_orphans = true` would
+        // silently disable local orphan parking, so we forbid it like the
+        // other zero-valued limits. Also note there is no upper bound relative
+        // to the overall orphan pool capacity
+        // (`config::DEFAULT_ORPHAN_POOL_CAPACITY`): a value above it makes the
+        // separate local cap non-binding, since the total pool limit evicts
+        // without regard to origin.
+        ensure!(
+            *self.local_orphan_pool_capacity > 0,
+            ConfigError::LocalOrphanPoolCapacityCannotBeZero
+        );
         Ok(())
     }
 }
@@ -213,13 +237,15 @@ impl From<RpcMempoolConfig> for MempoolConfig {
             max_cluster_tx_count,
             max_cluster_size_bytes,
             allow_local_orphans,
+            local_orphan_pool_capacity,
         } = value;
 
         Self {
             min_tx_relay_fee_rate: min_tx_relay_fee_rate.into(),
             max_cluster_tx_count: max_cluster_tx_count.into(),
             max_cluster_size_bytes: max_cluster_size_bytes.into(),
             allow_local_orphans: allow_local_orphans.into(),
+            local_orphan_pool_capacity: local_orphan_pool_capacity.into(),
         }
     }
 }
@@ -231,13 +257,21 @@ pub struct RpcMempoolConfig {
     pub max_cluster_tx_count: usize,
     pub max_cluster_size_bytes: usize,
     pub allow_local_orphans: bool,
+    pub local_orphan_pool_capacity: usize,
 }
 
+/// Mempool configuration errors.
+// The variant names intentionally share the `CannotBeZero` suffix: they all
+// describe zero-values that make no sense for the corresponding limit.
+#[allow(clippy::enum_variant_names)]
 #[derive(Debug, Clone, thiserror::Error, PartialEq, Eq)]
 pub enum ConfigError {
     #[error("Maximum cluster transaction count cannot be zero")]
     MaxClusterTxCountCannotBeZero,
 
+    #[error("Local orphan pool capacity cannot be zero")]
+    LocalOrphanPoolCapacityCannotBeZero,
+
     #[error("Maximum cluster size in bytes cannot be zero")]
     MaxClusterSizeBytesCannotBeZero,
 }

mempool/src/error/mod.rs

@@ -205,8 +205,8 @@ pub enum OrphanPoolError {
     #[error("Orphan pool full")]
     Full,
 
-    #[error("Local orphan capacity exceeded")]
-    LocalCapacityExceeded,
+    #[error("Local orphan capacity exceeded (limit: {0})")]
+    LocalCapacityExceeded(usize),
 
     #[error("Account nonces too distant, gap: {0}")]
     NonceGapTooLarge(u64),

mempool/src/pool/mod.rs

@@ -550,8 +550,14 @@ impl<M: MemoryUsageEstimator + ShallowClone> Mempool<M> {
 
                 // The rebuild above may have enqueued orphans for reprocessing (e.g. parked
                 // local-origin transactions whose parents have just been re-added from blocks).
-                // Pump the work queue now so they don't have to wait for an external trigger.
-                while self.has_work() {
+                // Pump a bounded number of work units now so they don't have to wait for an
+                // external trigger, while keeping per-event processing latency bounded; the
+                // rest stays in the work queue for the round-based scheduler.
+                const MAX_WORK_UNITS_PER_TIP: usize = 64;
+                for _ in 0..MAX_WORK_UNITS_PER_TIP {
+                    if !self.has_work() {
+                        break;
+                    }
                     self.perform_work_unit();
                 }
             }
@@ -749,10 +755,11 @@ impl<'a> TxFinalizer<'a> {
             // Keep local-origin orphans from crowding out remote peers' orphans. Resubmissions
             // of already-parked transactions bypass the capacity check so that they keep
             // yielding the idempotent `InOrphanPoolDuplicate` status instead of an error.
+            let local_capacity = *tx_pool.mempool_config().local_orphan_pool_capacity;
             if !self.orphan_pool.contains(transaction.tx_id())
-                && self.orphan_pool.local_len() >= config::DEFAULT_LOCAL_ORPHAN_POOL_CAPACITY
+                && self.orphan_pool.local_len() >= local_capacity
             {
-                return Err(OrphanPoolError::LocalCapacityExceeded);
+                return Err(OrphanPoolError::LocalCapacityExceeded(local_capacity));
             }
         }
 

mempool/src/pool/tests/basic.rs

@@ -560,6 +560,7 @@ async fn get_all_txs_in_insertion_order(#[case] seed: Seed) {
         .build();
     let mempool_config = MempoolConfig {
         allow_local_orphans: Default::default(),
+        local_orphan_pool_capacity: Default::default(),
         min_tx_relay_fee_rate: FeeRate::from_amount_per_kb(Amount::ZERO).into(),
         max_cluster_tx_count: Default::default(),
         max_cluster_size_bytes: Default::default(),

mempool/src/pool/tests/orphans.rs

@@ -238,6 +238,7 @@ async fn transaction_graph_subset_permutation(#[case] seed: Seed) {
 
     let mempool_config = MempoolConfig {
         allow_local_orphans: Default::default(),
+        local_orphan_pool_capacity: Default::default(),
         min_tx_relay_fee_rate: TEST_MIN_TX_RELAY_FEE_RATE.into(),
         // Make sure we don't hit the max cluster tx count limit.
         max_cluster_tx_count: num_txs.into(),
@@ -464,15 +465,28 @@ fn make_parking_mempool(
 ) -> (
     crate::pool::Mempool<crate::pool::memory_usage_estimator::StoreMemoryUsageEstimator>,
     Id<GenBlock>,
+) {
+    make_parking_mempool_with_config(rng, |_| {})
+}
+
+/// Same as [`make_parking_mempool`], but with local-orphan parking always
+/// enabled and the mempool config adjustable before the mempool is created.
+fn make_parking_mempool_with_config(
+    rng: &mut impl CryptoRng,
+    configure: impl FnOnce(&mut MempoolConfig),
+) -> (
+    crate::pool::Mempool<crate::pool::memory_usage_estimator::StoreMemoryUsageEstimator>,
+    Id<GenBlock>,
 ) {
     let tf = TestFramework::builder(rng).build();
-    let genesis_id = tf.genesis().get_id();
+    let genesis_id = tf.genesis().get_id().into();
 
     let mut config = create_mempool_config();
     config.allow_local_orphans = true.into();
+    configure(&mut config);
 
     let mempool = setup_with_chainstate_generic(tf.chainstate(), config, Default::default());
-    (mempool, genesis_id.into())
+    (mempool, genesis_id)
 }
 
 #[rstest]
@@ -543,9 +557,12 @@ async fn local_orphan_reprocessed_when_parent_arrives(
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn local_orphan_capacity_limit(#[case] seed: Seed) {
     let mut rng = make_seedable_rng(seed);
-    let (mut mempool, genesis_id) = make_parking_mempool(&mut rng);
-
-    let local_capacity = crate::config::DEFAULT_LOCAL_ORPHAN_POOL_CAPACITY;
+    // Use a non-default capacity so the test verifies that the configured
+    // limit is propagated into the error, not just the compile-time default.
+    let local_capacity = 5;
+    let (mut mempool, genesis_id) = make_parking_mempool_with_config(&mut rng, |config| {
+        config.local_orphan_pool_capacity = local_capacity.into();
+    });
 
     // A parent with `local_capacity + 1` outputs, so that each child conflicts with no other
     let tx0 = make_tx(
@@ -584,7 +601,9 @@ async fn local_orphan_capacity_limit(#[case] seed: Seed) {
     let res = mempool
         .add_transaction_with_origin(children[local_capacity].clone(), LocalTxOrigin::P2p.into());
     match res {
-        Err(Error::Orphan(OrphanPoolError::LocalCapacityExceeded)) => {}
+        Err(Error::Orphan(OrphanPoolError::LocalCapacityExceeded(limit))) => {
+            assert_eq!(limit, local_capacity);
+        }
         other => panic!("Expected LocalCapacityExceeded error, got {other:?}"),
     }
 }

mempool/src/pool/tests/relatives.rs

@@ -832,6 +832,7 @@ fn create_mempool(
 ) -> MempoolType {
     let mempool_config = MempoolConfig {
         allow_local_orphans: Default::default(),
+        local_orphan_pool_capacity: Default::default(),
         min_tx_relay_fee_rate: FeeRate::from_amount_per_kb(Amount::ZERO).into(),
         max_cluster_tx_count,
         max_cluster_size_bytes,

mempool/src/pool/tx_pool/tests/basic.rs

@@ -388,6 +388,7 @@ async fn tx_bigger_than_cluster_size(#[case] seed: Seed) -> anyhow::Result<()> {
     let tx_size = tx.encoded_size();
     let mempool_config = MempoolConfig {
         allow_local_orphans: Default::default(),
+        local_orphan_pool_capacity: Default::default(),
         min_tx_relay_fee_rate: TEST_MIN_TX_RELAY_FEE_RATE.into(),
         max_cluster_size_bytes: max_cluster_size.into(),
 
@@ -1421,6 +1422,7 @@ async fn no_empty_bags_in_indices(#[case] seed: Seed) -> anyhow::Result<()> {
 
     let mempool_config = MempoolConfig {
         allow_local_orphans: Default::default(),
+        local_orphan_pool_capacity: Default::default(),
         min_tx_relay_fee_rate: TEST_MIN_TX_RELAY_FEE_RATE.into(),
         // Make sure we don't hit the max cluster tx count limit.
         max_cluster_tx_count: (num_child_txs + 1).into(),
@@ -1557,6 +1559,7 @@ async fn accepted_tx_size(
 
     let mempool_config = MempoolConfig {
         allow_local_orphans: Default::default(),
+        local_orphan_pool_capacity: Default::default(),
         min_tx_relay_fee_rate: TEST_MIN_TX_RELAY_FEE_RATE.into(),
         max_cluster_size_bytes: max_cluster_size.into(),
 
@@ -1598,6 +1601,7 @@ async fn initial_values(#[case] seed: Seed) {
 
     let mempool_config = MempoolConfig {
         allow_local_orphans: Default::default(),
+        local_orphan_pool_capacity: Default::default(),
         min_tx_relay_fee_rate: FeeRate::from_atoms_per_kb(
             rng.random_range(100_000..100_000_000_000_000),
         )
@@ -1679,6 +1683,7 @@ fn stack_overflow_on_transaction_addition(
             tf.chainstate(),
             MempoolConfig {
                 allow_local_orphans: Default::default(),
+                local_orphan_pool_capacity: Default::default(),
                 min_tx_relay_fee_rate: FeeRate::from_amount_per_kb(Amount::ZERO).into(),
                 max_cluster_size_bytes: usize::MAX.into(),
                 max_cluster_tx_count: max_cluster_tx_count.into(),
@@ -1840,6 +1845,7 @@ async fn storage_capacity_shrinkage(#[case] seed: Seed) {
 
     let mempool_config = MempoolConfig {
         allow_local_orphans: Default::default(),
+        local_orphan_pool_capacity: Default::default(),
         min_tx_relay_fee_rate: min_tx_relay_fee_rate.into(),
         // Make sure we don't hit the max cluster tx count limit.
         max_cluster_tx_count: txs.len().into(),

mempool/src/pool/tx_pool/tests/tx_ids_by_score_and_ancestry.rs

@@ -50,6 +50,7 @@ async fn tx_ids_by_score_and_ancestry(#[case] seed: Seed) {
         tf.chainstate(),
         MempoolConfig {
             allow_local_orphans: Default::default(),
+            local_orphan_pool_capacity: Default::default(),
             min_tx_relay_fee_rate: FeeRate::from_amount_per_kb(Amount::ZERO).into(),
             max_cluster_tx_count: Default::default(),
             max_cluster_size_bytes: Default::default(),
@@ -218,6 +219,7 @@ fn stack_overflow_on_misconfigured_cluster_size_limit(#[case] seed: Seed) {
             tf.chainstate(),
             MempoolConfig {
                 allow_local_orphans: Default::default(),
+                local_orphan_pool_capacity: Default::default(),
                 min_tx_relay_fee_rate: FeeRate::from_amount_per_kb(Amount::ZERO).into(),
                 max_cluster_size_bytes: usize::MAX.into(),
                 max_cluster_tx_count: usize::MAX.into(),

mempool/src/pool/tx_pool/tests/utils.rs

@@ -48,6 +48,7 @@ pub fn create_mempool_config() -> MempoolConfig {
         max_cluster_tx_count: Default::default(),
         max_cluster_size_bytes: Default::default(),
         allow_local_orphans: false.into(),
+        local_orphan_pool_capacity: Default::default(),
     }
 }
 
@@ -385,6 +386,7 @@ pub fn setup_with_min_tx_relay_fee_rate(fee_rate: FeeRate) -> TxPool<StoreMemory
         max_cluster_tx_count: Default::default(),
         max_cluster_size_bytes: Default::default(),
         allow_local_orphans: false.into(),
+        local_orphan_pool_capacity: Default::default(),
     };
     let chainstate_interface = start_chainstate_with_config(Arc::clone(&chain_config));
     TxPool::new(