fix(zebrad): release the per-peer mempool slot on verification timeout

The mempool downloader's verification-timeout arm removed the `cancel_handles`
entry (the GHSA-65jj memory fix) with a bare `remove`, discarding the stored
`source` and skipping the shared per-peer slot release. After
`MAX_INBOUND_CONCURRENCY_PER_PEER` timed-out transactions from one source, that
source's `pending_per_peer` count stayed pinned at the cap with no tasks left,
so further queueing from it returned `FullQueue`.

Destructure the removed handle and call `release_peer_slot`, matching the
success and verifier-error arms while preserving the GHSA-65jj handle removal.
Adds a regression test that times out `MAX_INBOUND_CONCURRENCY_PER_PEER`
peer-sourced transactions and asserts a further one from the same source queues.

Author: natalieesk

CHANGELOG.md

@@ -43,6 +43,11 @@ and this project adheres to [Semantic Versioning](https://semver.org).
 
 ### Fixed
 
+- The mempool now releases a peer's per-peer download slot when a transaction verification times
+  out, instead of only on success or verifier error. Previously a peer whose transactions timed out
+  could pin its per-peer slots at the cap and be unable to queue any further transactions from that
+  address ([#10684](https://github.com/ZcashFoundation/zebra/issues/10684)).
+
 - `getblocksubsidy` now returns NU6-era funding stream metadata (recipient names and
   specification URLs) for NU6.1 and later upgrades. Amounts and addresses were never
   affected ([#11172](https://github.com/ZcashFoundation/zebra/pull/11172)).

zebrad/src/components/mempool/downloads.rs

@@ -265,12 +265,15 @@ where
                     (Ok(Err(Box::new((hash, e)))), Some(hash))
                 }
                 Err((txid, elapsed)) => {
-                    // Remove the cancel handle so the spawned task's queued `Gossip`
-                    // doesn't stay resident in `cancel_handles` after a verification
-                    // timeout. Without this, a peer that gets each transaction to
-                    // hit `RATE_LIMIT_DELAY` can leak ~2 MB per tx until OOM.
-                    this.cancel_handles.remove(&txid);
-                    (Err((txid, elapsed)), None)
+                    // Treat a verification timeout as a terminal completion so the
+                    // shared cleanup below removes the `cancel_handles` entry — the
+                    // GHSA-65jj fix, which drops the resident `Gossip` (~2 MB per tx)
+                    // so it can't leak until OOM — and releases the per-peer queue
+                    // slot. A bare handle removal here left the slot pinned, so a
+                    // peer that timed out `MAX_INBOUND_CONCURRENCY_PER_PEER`
+                    // transactions could no longer queue any from that source.
+                    // See #10684.
+                    (Err((txid, elapsed)), Some(txid))
                 }
             };
 

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

@@ -2284,3 +2284,85 @@ async fn cancel_handles_drained_after_verification_timeout() {
         "regression GHSA-65jj-fmw8-468q: cancel_handles must be drained after timeout"
     );
 }
+
+/// Regression test for #10684: the mempool verification-timeout path must
+/// release the per-peer queue slot, not just remove the cancel handle.
+///
+/// Before the fix, a peer whose `MAX_INBOUND_CONCURRENCY_PER_PEER` transactions
+/// all hit the verification timeout kept its per-peer count pinned at the cap
+/// even though no tasks remained, so any further transaction from that source
+/// was rejected with `FullQueue`.
+#[tokio::test(flavor = "current_thread", start_paused = true)]
+async fn verification_timeout_releases_peer_slot() {
+    use std::net::SocketAddr;
+
+    use futures::stream::StreamExt;
+    use tower::timeout::Timeout;
+    use zebra_node_services::mempool::Gossip;
+
+    use crate::components::mempool::{
+        crawler::RATE_LIMIT_DELAY,
+        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 source: SocketAddr = "127.0.0.1:8233".parse().expect("valid socket addr");
+
+    let mut iter = Network::Mainnet.unmined_transactions_in_blocks(1..=10);
+
+    // Fill the per-peer slot with `MAX_INBOUND_CONCURRENCY_PER_PEER` peer-sourced
+    // transactions from a single source.
+    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(source), None)
+            .unwrap_or_else(|e| panic!("queue tx {i} failed: {e:?}"));
+    }
+
+    assert_eq!(downloads.in_flight(), MAX_INBOUND_CONCURRENCY_PER_PEER);
+
+    // Advance past `RATE_LIMIT_DELAY` so every spawned task hits the verification
+    // timeout (the mocked services never make progress).
+    time::advance(RATE_LIMIT_DELAY + Duration::from_secs(5)).await;
+    tokio::task::yield_now().await;
+
+    for _ in 0..MAX_INBOUND_CONCURRENCY_PER_PEER {
+        match downloads.as_mut().next().await {
+            Some(Err(_)) => {}
+            Some(Ok(_)) => panic!("expected a verification timeout error"),
+            None => panic!("Downloads stream ended before all tasks resolved"),
+        }
+    }
+
+    assert_eq!(downloads.in_flight(), 0, "pending should be drained");
+    assert_eq!(
+        downloads.transaction_requests().count(),
+        0,
+        "cancel_handles must be drained after timeout (GHSA-65jj)"
+    );
+
+    // The per-peer slots must have been released: a further transaction from the
+    // same source must queue successfully. Before the fix this returned
+    // `FullQueue` because the timeout path never released the slots.
+    let tx = iter.next().expect("enough vector txs").transaction;
+    downloads
+        .as_mut()
+        .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");
+}