fix(zebrad): key the mempool per-peer download cap on IpAddr

CHANGELOG.md

@@ -68,6 +68,10 @@ and this project adheres to [Semantic Versioning](https://semver.org).
   peer's IPv4 address did not disconnect it while it stayed connected, and the same peer counted
   twice towards the per-IP inbound connection limit
   ([#10695](https://github.com/ZcashFoundation/zebra/issues/10695)).
+- The mempool per-peer inbound-download cap is now keyed on the peer's IP address instead of its
+  full socket address, so a single host can no longer exceed the cap by opening connections from
+  multiple source ports. This matches the inbound-block download cap
+  ([#10685](https://github.com/ZcashFoundation/zebra/issues/10685)).
 
 ## [Zebra 6.2.3](https://github.com/ZcashFoundation/zebra/releases/tag/v6.2.3) - 2026-07-27
 

zebrad/src/components/mempool/downloads.rs

@@ -27,7 +27,7 @@
 //! [`Mempool::poll_ready`]: super::Mempool::poll_ready
 use std::{
     collections::{HashMap, HashSet},
-    net::SocketAddr,
+    net::{IpAddr, SocketAddr},
     pin::Pin,
     task::{Context, Poll},
     time::Duration,
@@ -206,12 +206,19 @@ where
         ),
     >,
 
-    /// The number of currently in-flight download tasks per advertising peer.
+    /// The number of currently in-flight download tasks per advertising host,
+    /// keyed on the peer's [`IpAddr`].
     ///
-    /// Invariant: a peer is present here iff some entry in [`Self::cancel_handles`]
-    /// has it as the third tuple element. Enforces
+    /// Keyed on the host IP rather than the full `SocketAddr` so that all
+    /// connections from one host share a single budget: the `source` is a
+    /// transient `(IP, ephemeral port)`, so per-`SocketAddr` keying would give
+    /// each connection its own bucket. This matches the inbound-block download
+    /// cap (`in_flight_ips`). See #10685.
+    ///
+    /// Invariant: an IP is present here with count `n` iff exactly `n` entries in
+    /// [`Self::cancel_handles`] have a source with that IP. Enforces
     /// [`MAX_INBOUND_CONCURRENCY_PER_PEER`]. See `GHSA-4fc2-h7jh-287c`.
-    pending_per_peer: HashMap<SocketAddr, usize>,
+    pending_per_peer: HashMap<IpAddr, usize>,
 }
 
 impl<ZN, ZV, ZS> Stream for Downloads<ZN, ZV, ZS>
@@ -371,10 +378,10 @@ where
             return Err(MempoolError::FullQueue);
         }
 
-        // Per-peer cap: a single advertising peer cannot saturate the queue
-        // with attacker-supplied fake txids. See `GHSA-4fc2-h7jh-287c`.
+        // Per-peer cap: a single advertising host (keyed by IP) cannot saturate
+        // the queue with attacker-supplied fake txids. See `GHSA-4fc2-h7jh-287c`.
         if let Some(source) = source {
-            let count = self.pending_per_peer.get(&source).copied().unwrap_or(0);
+            let count = self.pending_per_peer.get(&source.ip()).copied().unwrap_or(0);
             if count >= MAX_INBOUND_CONCURRENCY_PER_PEER {
                 debug!(
                     ?txid,
@@ -550,7 +557,7 @@ where
         if let Some(source) = source {
             // The per-peer cap check above ensures this can't exceed
             // `MAX_INBOUND_CONCURRENCY_PER_PEER`.
-            *self.pending_per_peer.entry(source).or_insert(0) += 1;
+            *self.pending_per_peer.entry(source.ip()).or_insert(0) += 1;
         }
 
         debug!(
@@ -604,13 +611,14 @@ where
         metrics::gauge!("mempool.currently.queued.transactions",).set(self.pending.len() as f64);
     }
 
-    /// Decrement the per-peer pending count for `source`, removing the entry
+    /// Decrement the per-host pending count for `source`'s IP, removing the entry
     /// when it reaches zero.
-    fn release_peer_slot(pending_per_peer: &mut HashMap<SocketAddr, usize>, source: SocketAddr) {
-        if let Some(count) = pending_per_peer.get_mut(&source) {
+    fn release_peer_slot(pending_per_peer: &mut HashMap<IpAddr, usize>, source: SocketAddr) {
+        let ip = source.ip();
+        if let Some(count) = pending_per_peer.get_mut(&ip) {
             *count = count.saturating_sub(1);
             if *count == 0 {
-                pending_per_peer.remove(&source);
+                pending_per_peer.remove(&ip);
             }
         }
     }

zebrad/src/components/mempool/tests/vector.rs

@@ -2366,3 +2366,61 @@ async fn verification_timeout_releases_peer_slot() {
         .download_if_needed_and_verify(Gossip::Tx(tx), Some(source), None)
         .expect("a further transaction from the same source should queue after slots are freed");
 }
+
+/// Regression test for #10685: the mempool per-peer download cap must be keyed
+/// on the peer's `IpAddr`, not the full `SocketAddr`.
+///
+/// The `source` is a transient `(IP, ephemeral port)`, so keying on the whole
+/// `SocketAddr` gave every connection from one host its own budget. Two sources
+/// with the same IP and different ports must share a single
+/// `MAX_INBOUND_CONCURRENCY_PER_PEER` bucket.
+#[tokio::test(flavor = "current_thread")]
+async fn per_peer_cap_is_keyed_on_ip_not_socket_addr() {
+    use std::net::SocketAddr;
+
+    use tower::timeout::Timeout;
+    use zebra_node_services::mempool::Gossip;
+
+    use crate::components::mempool::downloads::{
+        Downloads, MAX_INBOUND_CONCURRENCY_PER_PEER, TRANSACTION_DOWNLOAD_TIMEOUT,
+        TRANSACTION_VERIFY_TIMEOUT,
+    };
+
+    let _init_guard = zebra_test::init();
+
+    let peer_set: MockPeerSet = MockService::build().for_unit_tests();
+    let state: MockService<zs::Request, zs::Response, PanicAssertion> =
+        MockService::build().for_unit_tests();
+    let tx_verifier: MockTxVerifier = MockService::build().for_unit_tests();
+
+    let mut downloads = Box::pin(Downloads::new(
+        Timeout::new(peer_set, TRANSACTION_DOWNLOAD_TIMEOUT),
+        Timeout::new(tx_verifier, TRANSACTION_VERIFY_TIMEOUT),
+        state,
+    ));
+
+    let mut iter = Network::Mainnet.unmined_transactions_in_blocks(1..=10);
+
+    // Fill the per-host budget from one `(IP, port)`.
+    let first: SocketAddr = "127.0.0.1:8233".parse().expect("valid socket addr");
+    for i in 0..MAX_INBOUND_CONCURRENCY_PER_PEER {
+        let tx = iter.next().expect("enough vector txs").transaction;
+        downloads
+            .as_mut()
+            .download_if_needed_and_verify(Gossip::Tx(tx), Some(first), None)
+            .unwrap_or_else(|e| panic!("queue tx {i} failed: {e:?}"));
+    }
+
+    // A further transaction from the SAME IP but a DIFFERENT port must be
+    // rejected, because the cap is keyed on the host IP. Before the fix it
+    // landed in a separate `SocketAddr` bucket and was accepted.
+    let tx = iter.next().expect("enough vector txs").transaction;
+    let same_ip_other_port: SocketAddr = "127.0.0.1:9999".parse().expect("valid socket addr");
+    let result = downloads
+        .as_mut()
+        .download_if_needed_and_verify(Gossip::Tx(tx), Some(same_ip_other_port), None);
+    assert!(
+        matches!(result, Err(MempoolError::FullQueue)),
+        "a transaction from the same IP on a different port must share the per-peer bucket, got {result:?}"
+    );
+}