fix: bound inbound p2p work handled per network pass

changelog.d/p2p-bound-inbound-work.added

@@ -0,0 +1 @@
+New `[connection_options]` settings bound how much inbound p2p work a node does in one network pass: `max_inbound_messages_per_pass` (default `64`) caps the messages handled from any one peer, and `max_inbound_processing_ms` (default `500`) caps the time spent handling inbound messages before the node moves on to inventory sync, block downloads and StackerDB sync. Work that doesn't fit is carried over to the next pass, which starts immediately unless the node is still processing previously downloaded blocks. While a peer's leftover messages are queued, the node stops reading from that peer's socket, so a burst of messages can't overflow the inbox and drop the connection.

stacks-node/src/nakamoto_node/peer.rs

@@ -252,6 +252,12 @@ impl PeerThread {
                 self.net.has_more_downloads()
             );
             1
+        } else if !download_backpressure && self.net.has_pending_inbound_work() {
+            // finish handling inbound messages left over from the last pass.  Like the
+            // downloader, don't spin while the relayer is still working through the blocks we
+            // already handed it -- the leftover messages stay queued until it catches up.
+            debug!("P2P: inbound messages pending");
+            1
         } else {
             self.poll_timeout
         };

stacks-node/src/neon_node.rs

@@ -4502,6 +4502,12 @@ impl PeerThread {
                 self.get_network().has_more_downloads()
             );
             1
+        } else if !download_backpressure && self.get_network().has_pending_inbound_work() {
+            // finish handling inbound messages left over from the last pass.  Like the
+            // downloader, don't spin while the relayer is still working through the blocks we
+            // already handed it -- the leftover messages stay queued until it catches up.
+            debug!("P2P: inbound messages pending");
+            1
         } else {
             self.poll_timeout
         };

stackslib/src/config/mod.rs

@@ -3885,6 +3885,22 @@ pub struct ConnectionOptionsFile {
     /// @default: `0` (disabled)
     /// @units: bytes/second
     pub max_nakamoto_block_push_bandwidth: Option<u64>,
+
+    /// Maximum number of inbound p2p messages handled from a single peer in one
+    /// network pass. Messages beyond this stay queued and are handled in the next
+    /// pass. Zero disables the cap.
+    /// ---
+    /// @default: `64`
+    pub max_inbound_messages_per_pass: Option<usize>,
+
+    /// Maximum time spent handling inbound p2p messages in one network pass,
+    /// before the node moves on to the rest of the pass (inventory sync, block
+    /// downloads, StackerDB sync, ...). Peers not handled in time are handled
+    /// first in the next pass. Zero disables the limit.
+    /// ---
+    /// @default: `500`
+    /// @units: milliseconds
+    pub max_inbound_processing_ms: Option<u64>,
 }
 
 impl ConnectionOptionsFile {
@@ -4052,6 +4068,12 @@ impl ConnectionOptionsFile {
             max_nakamoto_block_push_bandwidth: self
                 .max_nakamoto_block_push_bandwidth
                 .unwrap_or(default.max_nakamoto_block_push_bandwidth),
+            max_inbound_messages_per_pass: self
+                .max_inbound_messages_per_pass
+                .unwrap_or(default.max_inbound_messages_per_pass),
+            max_inbound_processing_ms: self
+                .max_inbound_processing_ms
+                .unwrap_or(default.max_inbound_processing_ms),
             ..default
         })
     }
@@ -6052,6 +6074,45 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_load_inbound_limit_fields_config() {
+        // check defaults for omitted fields
+        let config = utils::config_from_valid_string(
+            r#"
+            [burnchain]
+            mode = "krypton"
+            "#,
+        );
+        assert_eq!(64, config.connection_options.max_inbound_messages_per_pass);
+        assert_eq!(500, config.connection_options.max_inbound_processing_ms);
+
+        // check values for configured fields
+        let config = utils::config_from_valid_string(
+            r#"
+            [burnchain]
+            mode = "krypton"
+            [connection_options]
+            max_inbound_messages_per_pass = 10
+            max_inbound_processing_ms = 20
+            "#,
+        );
+        assert_eq!(10, config.connection_options.max_inbound_messages_per_pass);
+        assert_eq!(20, config.connection_options.max_inbound_processing_ms);
+
+        // zero disables the limits
+        let config = utils::config_from_valid_string(
+            r#"
+            [burnchain]
+            mode = "krypton"
+            [connection_options]
+            max_inbound_messages_per_pass = 0
+            max_inbound_processing_ms = 0
+            "#,
+        );
+        assert_eq!(0, config.connection_options.max_inbound_messages_per_pass);
+        assert_eq!(0, config.connection_options.max_inbound_processing_ms);
+    }
+
     /// There are three different ways to start a node with a default config:
     ///
     /// - via `stacks-node mainnet`

stackslib/src/net/chat.rs

@@ -17,6 +17,8 @@
 use std::collections::{HashMap, HashSet, VecDeque};
 use std::io::{Read, Write};
 use std::net::SocketAddr;
+use std::num::NonZeroUsize;
+use std::time::Instant;
 use std::{cmp, mem};
 
 use clarity::vm::types::QualifiedContractIdentifier;
@@ -539,6 +541,33 @@ impl Neighbor {
     }
 }
 
+/// How much inbound work `ConversationP2P::chat_within()` may do in one network pass
+#[derive(Debug, Clone, Copy, PartialEq)]
+pub struct InboundBudget {
+    /// most inbox messages to handle; `None` means no limit
+    pub max_messages: Option<NonZeroUsize>,
+    /// stop handling inbox messages once this passes; `None` means no limit
+    pub deadline: Option<Instant>,
+}
+
+impl InboundBudget {
+    /// Budget from a `max_inbound_messages_per_pass`-style cap, where zero means no limit
+    pub fn new(max_messages: usize, deadline: Option<Instant>) -> Self {
+        Self {
+            max_messages: NonZeroUsize::new(max_messages),
+            deadline,
+        }
+    }
+
+    /// No limits: handle everything in the inbox
+    pub fn unbounded() -> Self {
+        Self {
+            max_messages: None,
+            deadline: None,
+        }
+    }
+}
+
 impl ConversationP2P {
     /// Create an unconnected conversation
     pub fn new(
@@ -2908,6 +2937,7 @@ impl ConversationP2P {
     /// Carry on a conversation with the remote peer.
     /// Called from the p2p network thread, so no need for a network handle.
     /// Attempts to fulfill requests in other threads as a result of processing a message.
+    /// Handles at most `max_inbound_messages_per_pass` inbox messages; the rest stay queued.
     /// Returns the list of unfulfilled Stacks messages we received -- messages not destined for
     /// any other thread in this program (i.e. "unsolicited messages").
     pub fn chat(
@@ -2918,11 +2948,54 @@ impl ConversationP2P {
         dns_client_opt: &mut Option<&mut DNSClient>,
         ibd: bool,
     ) -> Result<Vec<StacksMessage>, net_error> {
+        let budget =
+            InboundBudget::new(self.connection.options.max_inbound_messages_per_pass, None);
+        self.chat_within(network, sortdb, chainstate, dns_client_opt, ibd, budget)
+    }
+
+    /// Like `chat()`, but handles inbox messages within `budget` instead: at most
+    /// `budget.max_messages` of them, stopping once `budget.deadline` has passed so one peer's
+    /// batch can't run the network pass past its `max_inbound_processing_ms` budget.  Always
+    /// handles at least one message so the conversation makes progress.
+    pub fn chat_within(
+        &mut self,
+        network: &mut PeerNetwork,
+        sortdb: &SortitionDB,
+        chainstate: &mut StacksChainState,
+        dns_client_opt: &mut Option<&mut DNSClient>,
+        ibd: bool,
+        budget: InboundBudget,
+    ) -> Result<Vec<StacksMessage>, net_error> {
+        let InboundBudget {
+            max_messages,
+            deadline,
+        } = budget;
         let num_inbound = self.connection.inbox_len();
         debug!("{:?}: {} messages pending", &self, num_inbound);
 
+        // don't let one peer monopolize the p2p thread -- leave the rest of its messages queued
+        // for the next pass
+        let num_to_handle =
+            max_messages.map_or(num_inbound, |max| cmp::min(num_inbound, max.get()));
+        if num_to_handle < num_inbound {
+            debug!(
+                "{:?}: handle {} of {} pending messages this pass",
+                &self, num_to_handle, num_inbound
+            );
+        }
+
         let mut unsolicited = vec![];
-        for _ in 0..num_inbound {
+        for num_handled in 0..num_to_handle {
+            if num_handled > 0 && deadline.is_some_and(|deadline| Instant::now() >= deadline) {
+                debug!(
+                    "{:?}: out of time after handling {} messages; leaving {} queued for the next pass",
+                    &self,
+                    num_handled,
+                    self.connection.inbox_len()
+                );
+                break;
+            }
+
             let update_stats; // whether or not this message can count towards this peer's liveness stats
             let mut msg = match self.connection.next_inbox_message() {
                 None => {
@@ -4789,6 +4862,168 @@ mod test {
         }
     }
 
+    #[test]
+    fn convo_chat_inbound_limits() {
+        let conn_opts = ConnectionOptions::default();
+        let conn_opts_2 = ConnectionOptions {
+            max_inbound_messages_per_pass: 2,
+            ..ConnectionOptions::default()
+        };
+        let socketaddr_1 = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 8080);
+        let socketaddr_2 = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(1, 2, 3, 4)), 8081);
+
+        let mut chain_view = BurnchainView {
+            burn_block_height: 12348,
+            burn_block_hash: BurnchainHeaderHash([0x11; 32]),
+            burn_stable_block_height: 12341,
+            burn_stable_block_hash: BurnchainHeaderHash([0x22; 32]),
+            last_burn_block_hashes: HashMap::new(),
+            rc_consensus_hash: ConsensusHash([0x33; 20]),
+        };
+        chain_view.make_test_data();
+
+        let test_name_1 = "convo_chat_inbound_limits_1";
+        let test_name_2 = "convo_chat_inbound_limits_2";
+
+        let burnchain_1 = testing_burnchain_config(test_name_1);
+        let burnchain_2 = testing_burnchain_config(test_name_2);
+
+        let (mut peerdb_1, mut sortdb_1, _stackerdbs_1, _pox_id_1, mut chainstate_1) =
+            make_test_chain_dbs(
+                test_name_1,
+                &burnchain_1,
+                0x9abcdef0,
+                12350,
+                UrlString::from_literal("http://peer1.com"),
+                &[],
+                &[],
+                DEFAULT_SERVICES,
+            );
+        let (mut peerdb_2, mut sortdb_2, _stackerdbs_2, _pox_id_2, mut chainstate_2) =
+            make_test_chain_dbs(
+                test_name_2,
+                &burnchain_2,
+                0x9abcdef0,
+                12351,
+                UrlString::from_literal("http://peer2.com"),
+                &[],
+                &[],
+                DEFAULT_SERVICES,
+            );
+
+        let mut net_1 = db_setup(
+            test_name_1,
+            &burnchain_1,
+            0x9abcdef0,
+            &mut peerdb_1,
+            &mut sortdb_1,
+            &socketaddr_1,
+            &chain_view,
+        );
+        let mut net_2 = db_setup(
+            test_name_2,
+            &burnchain_2,
+            0x9abcdef0,
+            &mut peerdb_2,
+            &mut sortdb_2,
+            &socketaddr_2,
+            &chain_view,
+        );
+
+        let local_peer_1 = PeerDB::get_local_peer(peerdb_1.conn()).unwrap();
+
+        let mut convo_1 = ConversationP2P::new(
+            123,
+            456,
+            &burnchain_1,
+            &socketaddr_2,
+            &conn_opts,
+            true,
+            0,
+            StacksEpoch::unit_test_up_to(0, StacksEpochId::Epoch20),
+        );
+        let mut convo_2 = ConversationP2P::new(
+            123,
+            456,
+            &burnchain_2,
+            &socketaddr_1,
+            &conn_opts_2,
+            true,
+            0,
+            StacksEpoch::unit_test_up_to(0, StacksEpochId::Epoch20),
+        );
+
+        // convo_1 sends a handshake and three pings to convo_2
+        let handshake_1 = convo_1
+            .sign_message(
+                &chain_view,
+                &local_peer_1.private_key,
+                StacksMessageType::Handshake(HandshakeData::from_local_peer(&local_peer_1)),
+            )
+            .unwrap();
+        let mut rh_handshake_1 = convo_1.send_signed_request(handshake_1, 1000000).unwrap();
+
+        let mut pings = vec![];
+        let mut rh_pings = vec![];
+        for _ in 0..3 {
+            let ping_data = PingData::new();
+            let ping = convo_1
+                .sign_message(
+                    &chain_view,
+                    &local_peer_1.private_key,
+                    StacksMessageType::Ping(ping_data.clone()),
+                )
+                .unwrap();
+            rh_pings.push(convo_1.send_signed_request(ping, 1000000).unwrap());
+            pings.push(ping_data);
+        }
+
+        let mut send_handles: Vec<_> = rh_pings.iter_mut().collect();
+        send_handles.push(&mut rh_handshake_1);
+        convo_send_recv(&mut convo_1, send_handles, &mut convo_2);
+        assert_eq!(convo_2.connection.inbox_len(), 4);
+
+        // with the pass deadline already expired, convo_2 still makes progress on one message
+        convo_2
+            .chat_within(
+                &mut net_2,
+                &sortdb_2,
+                &mut chainstate_2,
+                &mut None,
+                false,
+                InboundBudget::new(2, Some(Instant::now())),
+            )
+            .unwrap();
+        assert_eq!(convo_2.connection.inbox_len(), 3);
+
+        // otherwise it handles two messages per pass, leaving the rest queued
+        convo_2
+            .chat(&mut net_2, &sortdb_2, &mut chainstate_2, &mut None, false)
+            .unwrap();
+        assert_eq!(convo_2.connection.inbox_len(), 1);
+
+        convo_2
+            .chat(&mut net_2, &sortdb_2, &mut chainstate_2, &mut None, false)
+            .unwrap();
+        assert_eq!(convo_2.connection.inbox_len(), 0);
+
+        // every ping was answered
+        let mut recv_handles: Vec<_> = rh_pings.iter_mut().collect();
+        recv_handles.push(&mut rh_handshake_1);
+        convo_send_recv(&mut convo_2, recv_handles, &mut convo_1);
+        convo_1
+            .chat(&mut net_1, &sortdb_1, &mut chainstate_1, &mut None, false)
+            .unwrap();
+
+        for (rh_ping, ping_data) in rh_pings.into_iter().zip(pings.iter()) {
+            let reply = rh_ping.recv(0).unwrap();
+            let StacksMessageType::Pong(data) = reply.payload else {
+                panic!("Unexpected payload message type");
+            };
+            assert_eq!(data.nonce, ping_data.nonce);
+        }
+    }
+
     #[test]
     fn convo_handshake_ping_loop() {
         let conn_opts = ConnectionOptions::default();