Merge bitcoin/bitcoin#34743: p2p: don't disconnect manual peers for block stalling

### doc/release-notes-34743.md
@@ -0,0 +1,7 @@
+P2P and network changes
+-----------------------
+
+- Manual peers (added using `addnode` or `-connect`) are no longer disconnected
+  during IBD for block stalling. Instead, their in-flight block requests are
+  released so other peers can continue block download progress, and the manual
+  peer is temporarily skipped for block downloads.

### src/net.cpp
@@ -1898,9 +1898,11 @@ bool CConnman::AddConnection(const std::string& address, ConnectionType conn_typ
     std::optional<int> max_connections;
     switch (conn_type) {
     case ConnectionType::INBOUND:
-    case ConnectionType::MANUAL:
     case ConnectionType::PRIVATE_BROADCAST:
         return false;
+    // no separate per-type limit for MANUAL because semAddnode limits them
+    case ConnectionType::MANUAL:
+        break;
     case ConnectionType::OUTBOUND_FULL_RELAY:
         max_connections = m_max_outbound_full_relay;
         break;
@@ -1922,8 +1924,8 @@ bool CConnman::AddConnection(const std::string& address, ConnectionType conn_typ
     // Max connections of specified type already exist
     if (max_connections != std::nullopt && existing_connections >= max_connections) return false;
 
-    // Max total outbound connections already exist
-    CountingSemaphoreGrant<> grant(*semOutbound, true);
+    // Max total automatic outbound or manual connections already exist
+    CountingSemaphoreGrant<> grant(conn_type == ConnectionType::MANUAL ? *semAddnode : *semOutbound, true);
     if (!grant) return false;
 
     OpenNetworkConnection(/*addrConnect=*/CAddress{},

### src/net.h
@@ -1368,13 +1368,13 @@ class CConnman
      * Attempts to open a connection. Currently only used from tests.
      *
      * @param[in]   address     Address of node to try connecting to
-     * @param[in]   conn_type   ConnectionType::OUTBOUND, ConnectionType::BLOCK_RELAY,
-     *                          ConnectionType::ADDR_FETCH or ConnectionType::FEELER
+     * @param[in]   conn_type   ConnectionType::OUTBOUND_FULL_RELAY, ConnectionType::BLOCK_RELAY,
+     *                          ConnectionType::ADDR_FETCH, ConnectionType::FEELER or ConnectionType::MANUAL
      * @param[in]   use_v2transport  Set to true if node attempts to connect using BIP 324 v2 transport protocol.
      * @return      bool        Returns false if there are no available
      *                          slots for this connection:
      *                          - conn_type not a supported ConnectionType
-     *                          - Max total outbound connection capacity filled
+     *                          - Max total automatic outbound or manual connection capacity filled
      *                          - Max connection capacity for type is filled
      */
     bool AddConnection(const std::string& address, ConnectionType conn_type, bool use_v2transport)

### src/net_processing.cpp
@@ -136,6 +136,8 @@ static const int MAX_BLOCKS_IN_TRANSIT_PER_PEER = 16;
 static constexpr auto BLOCK_STALLING_TIMEOUT_DEFAULT{2s};
 /** Maximum timeout for stalling block download. */
 static constexpr auto BLOCK_STALLING_TIMEOUT_MAX{64s};
+/** Time to avoid requesting blocks from a manual peer after it stalls block download. */
+static constexpr auto MANUAL_PEER_BLOCK_DOWNLOAD_COOLDOWN{2min};
 /** Maximum depth of blocks we're willing to serve as compact blocks to peers
  *  when requested. For older blocks, a regular BLOCK response will be sent. */
 static const int MAX_CMPCTBLOCK_DEPTH = 5;
@@ -454,6 +456,8 @@ struct CNodeState {
     std::list<QueuedBlock> vBlocksInFlight;
     //! When the first entry in vBlocksInFlight started downloading. Don't care when vBlocksInFlight is empty.
     std::chrono::microseconds m_downloading_since{0us};
+    //! Time before which block requests should not be sent to this peer.
+    std::chrono::microseconds m_block_download_paused_until{0us};
     //! Whether we consider this a preferred download peer.
     bool fPreferredDownload{false};
     /** Whether this peer wants invs or cmpctblocks (when possible) for block announcements. */
@@ -6438,10 +6442,18 @@ bool PeerManagerImpl::SendMessages(CNode& node)
             // Stalling only triggers when the block download window cannot move. During normal steady state,
             // the download window should be much larger than the to-be-downloaded set of blocks, so disconnection
             // should only happen during initial block download.
-            LogInfo("Peer is stalling block download, %s", node.DisconnectMsg());
-            node.fDisconnect = true;
-            // Increase timeout for the next peer so that we don't disconnect multiple peers if our own
-            // bandwidth is insufficient.
+            if (node.IsManualConn()) {
+                LogInfo("Pausing block downloads from stalling manual peer=%d for %d seconds\n", node.GetId(), count_seconds(MANUAL_PEER_BLOCK_DOWNLOAD_COOLDOWN));
+                state.m_block_download_paused_until = current_time + MANUAL_PEER_BLOCK_DOWNLOAD_COOLDOWN;
+                while (!state.vBlocksInFlight.empty()) {
+                    RemoveBlockRequest(state.vBlocksInFlight.front().pindex->GetBlockHash(), node.GetId());
+                }
+            } else {
+                LogInfo("Peer is stalling block download, %s", node.DisconnectMsg());
+                node.fDisconnect = true;
+            }
+            // Increase the timeout for the next peer so that we don't repeatedly react to apparent
+            // stalls caused by insufficient local bandwidth.
             const auto new_timeout = std::min(2 * stalling_timeout, BLOCK_STALLING_TIMEOUT_MAX);
             if (stalling_timeout != new_timeout && m_block_stalling_timeout.compare_exchange_strong(stalling_timeout, new_timeout)) {
                 LogDebug(BCLog::NET, "Increased stalling timeout temporarily to %d seconds\n", count_seconds(new_timeout));
@@ -6503,7 +6515,8 @@ bool PeerManagerImpl::SendMessages(CNode& node)
         // Message: getdata (blocks)
         //
         std::vector<CInv> vGetData;
-        if (CanServeBlocks(peer) && ((sync_blocks_and_headers_from_peer && !IsLimitedPeer(peer)) || !m_chainman.IsInitialBlockDownload()) && state.vBlocksInFlight.size() < MAX_BLOCKS_IN_TRANSIT_PER_PEER) {
+        const bool can_request_blocks_from_peer{current_time >= state.m_block_download_paused_until};
+        if (CanServeBlocks(peer) && can_request_blocks_from_peer && ((sync_blocks_and_headers_from_peer && !IsLimitedPeer(peer)) || !m_chainman.IsInitialBlockDownload()) && state.vBlocksInFlight.size() < MAX_BLOCKS_IN_TRANSIT_PER_PEER) {
             std::vector<const CBlockIndex*> vToDownload;
             NodeId staller = -1;
             auto get_inflight_budget = [&state]() {

### src/rpc/net.cpp
@@ -343,8 +343,9 @@ static RPCMethod addnode()
         "addnode",
         "Attempts to add or remove a node from the addnode list.\n"
                 "Or try a connection to a node once.\n"
-                "Nodes added using addnode (or -connect) are protected from DoS disconnection and are not required to be\n"
-                "full nodes/support SegWit as other outbound peers are (though such peers will not be synced from).\n" +
+                "Nodes added using addnode (or -connect) are protected from DoS disconnection and IBD block stalling\n"
+                "disconnection, and are not required to be full nodes or support SegWit as other outbound peers are (though\n"
+                "such peers will not be synced from).\n" +
                 strprintf("Addnode connections are limited to %u at a time", MAX_ADDNODE_CONNECTIONS) +
                 " and are counted separately from the -maxconnections limit.\n",
                 {
@@ -419,7 +420,7 @@ static RPCMethod addconnection()
         "Open an outbound connection to a specified node. This RPC is for testing only.\n",
         {
             {"address", RPCArg::Type::STR, RPCArg::Optional::NO, "The IP address and port to attempt connecting to."},
-            {"connection_type", RPCArg::Type::STR, RPCArg::Optional::NO, "Type of connection to open (\"outbound-full-relay\", \"block-relay-only\", \"addr-fetch\" or \"feeler\")."},
+            {"connection_type", RPCArg::Type::STR, RPCArg::Optional::NO, "Type of connection to open (\"outbound-full-relay\", \"block-relay-only\", \"addr-fetch\", \"feeler\" or \"manual\")."},
             {"v2transport", RPCArg::Type::BOOL, RPCArg::Optional::NO, "Attempt to connect using BIP324 v2 transport protocol"},
         },
         RPCResult{
@@ -449,6 +450,8 @@ static RPCMethod addconnection()
         conn_type = ConnectionType::ADDR_FETCH;
     } else if (conn_type_in == "feeler") {
         conn_type = ConnectionType::FEELER;
+    } else if (conn_type_in == "manual") {
+        conn_type = ConnectionType::MANUAL;
     } else {
         throw JSONRPCError(RPC_INVALID_PARAMETER, self.ToString());
     }

### test/functional/p2p_add_connections.py
@@ -29,13 +29,29 @@ def on_version(self, message):
 
 class P2PAddConnections(BitcoinTestFramework):
     def set_test_params(self):
-        self.num_nodes = 2
+        self.num_nodes = 3
+        self.bind_to_localhost_only = False
+        self.extra_args = [
+            ["-bind=127.0.0.1"],
+            ["-bind=127.0.0.1"],
+            ["-maxconnections=1", "-listen=0"],
+        ]
 
     def setup_network(self):
         self.setup_nodes()
         # Don't connect the nodes
 
     def run_test(self):
+        self.log.info("Add a manual connection after filling automatic outbound capacity")
+        self.nodes[2].add_outbound_p2p_connection(P2PInterface(), p2p_idx=0)
+        self.nodes[2].add_outbound_p2p_connection(
+            P2PInterface(), p2p_idx=1, connection_type="manual")
+        assert_equal(
+            {peer["connection_type"] for peer in self.nodes[2].getpeerinfo()},
+            {"manual", "outbound-full-relay"},
+        )
+        self.nodes[2].disconnect_p2ps()
+
         self.log.info("Add 8 outbounds to node 0")
         for i in range(8):
             self.log.info(f"outbound: {i}")

### test/functional/p2p_ibd_stalling.py
@@ -49,6 +49,10 @@ def set_test_params(self):
         self.num_nodes = 1
 
     def run_test(self):
+        self.test_stalling()
+        self.test_manual_peer_stalling()
+
+    def test_stalling(self):
         NUM_BLOCKS = 1025
         NUM_PEERS = 5
         node = self.nodes[0]
@@ -144,6 +148,93 @@ def run_test(self):
         self.log.info("Check that all outstanding blocks up to the second stall block get connected")
         self.wait_until(lambda: node.getblockcount() == second_stall_index)
 
+    def test_manual_peer_stalling(self):
+        self.log.info("Test that a manual peer is paused but not disconnected for stalling block download")
+        BLOCK_DOWNLOAD_COOLDOWN = 2 * 60
+        NUM_BLOCKS = 1025
+        node = self.nodes[0]
+        self.restart_node(0)
+        tip = int(node.getbestblockhash(), 16)
+        blocks = []
+        initial_height = node.getblockcount()
+        height = initial_height + 1
+        block_time = node.getblock(node.getbestblockhash())['time'] + 1
+        block_dict = {}
+        for _ in range(NUM_BLOCKS):
+            blocks.append(create_block(tip, height=height, ntime=block_time))
+            blocks[-1].solve()
+            tip = blocks[-1].hash_int
+            block_time += 1
+            height += 1
+            block_dict[blocks[-1].hash_int] = blocks[-1]
+        stall_block = blocks[0].hash_int
+
+        headers_message = msg_headers()
+        headers_message.headers = [CBlockHeader(b) for b in blocks]
+
+        self.mocktime = int(time.time()) + 1
+        node.setmocktime(self.mocktime)
+
+        self.log.info("Add a manual peer that stalls on block 0")
+        manual_peer = node.add_outbound_p2p_connection(
+            P2PStaller([stall_block]), p2p_idx=0, connection_type="manual")
+        manual_peer.block_store = block_dict
+        assert_equal(node.getpeerinfo()[0]['connection_type'], 'manual')
+
+        # Send headers to manual peer first so it gets block 0 assigned.
+        manual_peer.send_and_ping(headers_message)
+
+        self.log.info("Add outbound peers that serve all blocks")
+        outbound_peers = []
+        for i in range(4):
+            p = node.add_outbound_p2p_connection(
+                P2PStaller([]), p2p_idx=i + 1, connection_type="outbound-full-relay")
+            p.block_store = block_dict
+            p.send_and_ping(headers_message)
+            outbound_peers.append(p)
+
+        all_peers = [manual_peer] + outbound_peers
+
+        self.log.info("Wait until only the stall block remains in flight from the manual peer")
+        self.wait_until(lambda: sum(len(peer['inflight']) for peer in node.getpeerinfo()) == 1)
+        self.all_sync_send_with_ping(all_peers)
+        assert_equal(manual_peer.getdata_requests.count(stall_block), 1)
+        assert_equal(self.is_block_requested(outbound_peers, stall_block), False)
+
+        self.log.info("Advance time past stalling timeout and pause block downloads from the manual peer")
+        with node.assert_debug_log(["Pausing block downloads from stalling manual peer"]):
+            self.mocktime += 3
+            node.setmocktime(self.mocktime)
+            manual_peer.sync_with_ping()
+
+        assert_equal(manual_peer.is_connected, True)
+        assert_equal(node.num_test_p2p_connections(), len(all_peers))
+        assert_equal(manual_peer.getdata_requests.count(stall_block), 1)
+        assert_equal(sum(len(peer['inflight']) for peer in node.getpeerinfo() if peer['connection_type'] == 'manual'), 0)
+
+        self.log.info("Verify the released block is assigned to another peer")
+        self.all_sync_send_with_ping(outbound_peers)
+        self.wait_until(lambda: self.is_block_requested(outbound_peers, stall_block))
+
+        self.log.info("Verify that IBD completes while the manual peer is paused")
+        self.wait_until(lambda: node.getblockcount() == NUM_BLOCKS + initial_height)
+
+        self.log.info("Verify the manual peer is not assigned blocks during the cooldown")
+        post_cooldown_block = create_block(tip, height=height, ntime=block_time)
+        post_cooldown_block.solve()
+        manual_peer.block_store[post_cooldown_block.hash_int] = post_cooldown_block
+        post_cooldown_headers = msg_headers()
+        post_cooldown_headers.headers = [CBlockHeader(post_cooldown_block)]
+        manual_peer.send_and_ping(post_cooldown_headers)
+        assert_equal(post_cooldown_block.hash_int in manual_peer.getdata_requests, False)
+
+        self.log.info("Verify the manual peer can request blocks after the cooldown")
+        self.mocktime += BLOCK_DOWNLOAD_COOLDOWN + 1
+        node.setmocktime(self.mocktime)
+        manual_peer.sync_with_ping()
+        self.wait_until(lambda: post_cooldown_block.hash_int in manual_peer.getdata_requests)
+        self.wait_until(lambda: node.getblockcount() == NUM_BLOCKS + initial_height + 1)
+
 
     def all_sync_send_with_ping(self, peers):
         for p in peers:

### test/functional/test_framework/test_node.py
@@ -803,7 +803,7 @@ def add_p2p_connection(self, p2p_conn, *, wait_for_verack=True, send_version=Tru
 
     def add_outbound_p2p_connection(self, p2p_conn, *, wait_for_verack=True, wait_for_disconnect=False, p2p_idx, connection_type="outbound-full-relay", supports_v2_p2p=None, advertise_v2_p2p=None, **kwargs):
         """Add an outbound p2p connection from node. Must be an
-        "outbound-full-relay", "block-relay-only", "addr-fetch" or "feeler" connection.
+        "outbound-full-relay", "block-relay-only", "addr-fetch", "feeler" or "manual" connection.
 
         This method adds the p2p connection to the self.p2ps list and returns
         the connection to the caller.