From b2d1710433cb9f38c4e64856bd7b8dd9d4361501 Mon Sep 17 00:00:00 2001 From: Adesh Kolte Date: Tue, 28 Apr 2026 12:07:09 +0530 Subject: [PATCH 1/3] gossipsub: bound IWANT/IDONTWANT message_ids at decode time The truncate calls at behaviour.rs:3330 and 3347 only run after GossipsubCodec has already decoded the full Vec from the wire. A peer in the mesh can pack a single 10 MB+ frame with hundreds of thousands of IDs (~21 MB allocated for 477k 20-byte IDs) and Lighthouse will allocate them all before truncating to 5000. Recoverable, not a crash. Just allocator churn under sustained fire from a peer that's already in the mesh. max_messages_per_rpc caps the number of ControlAction entries, not IDs inside each one. Fix moves the cap to the codec by plumbing max_iwant_length and max_idontwant_messages from Config through ProtocolConfig into GossipsubCodec, then applying .take() before .collect() in the IWANT and IDONTWANT decode paths. The behaviour-layer truncate becomes a no-op for compliant inputs and remains as a safety net. Reported privately to security@sigmaprime.io; Kirk Baird approved public PR. --- .../src/behaviour/tests/topic_config.rs | 4 +++ protocols/gossipsub/src/config.rs | 2 ++ protocols/gossipsub/src/protocol.rs | 32 +++++++++++++++++-- 3 files changed, 36 insertions(+), 2 deletions(-) diff --git a/protocols/gossipsub/src/behaviour/tests/topic_config.rs b/protocols/gossipsub/src/behaviour/tests/topic_config.rs index ff14c6f0d01..7d5ae4bc772 100644 --- a/protocols/gossipsub/src/behaviour/tests/topic_config.rs +++ b/protocols/gossipsub/src/behaviour/tests/topic_config.rs @@ -667,6 +667,8 @@ fn test_validation_error_message_size_too_large_topic_specific() { Config::default_max_transmit_size() * 2, ValidationMode::None, max_transmit_size_map, + 5000, + 1000, ); let mut buf = BytesMut::new(); let rpc = proto::RPC { @@ -771,6 +773,8 @@ fn test_validation_message_size_within_topic_specific() { Config::default_max_transmit_size() * 2, ValidationMode::None, max_transmit_size_map, + 5000, + 1000, ); let mut buf = BytesMut::new(); let rpc = proto::RPC { diff --git a/protocols/gossipsub/src/config.rs b/protocols/gossipsub/src/config.rs index 5ca26ca35a6..9450f1a107f 100644 --- a/protocols/gossipsub/src/config.rs +++ b/protocols/gossipsub/src/config.rs @@ -1066,12 +1066,14 @@ impl ConfigBuilder { /// The maximum number of `message_ids` that we accept in a single IDONTWANT message. pub fn max_idontwant_messages(&mut self, size: usize) -> &mut Self { self.config.max_idontwant_messages = size; + self.config.protocol.max_idontwant_messages = size; self } /// The maximum number of `message_ids` that we accept in a single IWANT message. pub fn max_iwant_messages(&mut self, size: usize) -> &mut Self { self.config.max_iwant_length = size; + self.config.protocol.max_iwant_length = size; self } diff --git a/protocols/gossipsub/src/protocol.rs b/protocols/gossipsub/src/protocol.rs index 74dcc669f55..70dd2cd7039 100644 --- a/protocols/gossipsub/src/protocol.rs +++ b/protocols/gossipsub/src/protocol.rs @@ -72,6 +72,13 @@ pub struct ProtocolConfig { pub(crate) default_max_transmit_size: usize, /// The max transmit sizes for a topic. pub(crate) max_transmit_sizes: HashMap, + /// Cap on `message_ids` decoded per IWANT control message. Mirrored from + /// `Config::max_iwant_length` so the codec can bound allocations before + /// the full RPC reaches the behaviour layer. + pub(crate) max_iwant_length: usize, + /// Cap on `message_ids` decoded per IDONTWANT control message. Mirrored + /// from `Config::max_idontwant_messages`. + pub(crate) max_idontwant_messages: usize, } impl Default for ProtocolConfig { @@ -85,6 +92,8 @@ impl Default for ProtocolConfig { ], default_max_transmit_size: 65536, max_transmit_sizes: HashMap::new(), + max_iwant_length: 5000, + max_idontwant_messages: 1000, } } } @@ -139,6 +148,8 @@ where self.default_max_transmit_size, self.validation_mode, self.max_transmit_sizes, + self.max_iwant_length, + self.max_idontwant_messages, ), ), protocol_id.kind, @@ -162,6 +173,8 @@ where self.default_max_transmit_size, self.validation_mode, self.max_transmit_sizes, + self.max_iwant_length, + self.max_idontwant_messages, ), ), protocol_id.kind, @@ -178,6 +191,10 @@ pub struct GossipsubCodec { codec: quick_protobuf_codec::Codec, /// Maximum transmit sizes per topic, with a default if not specified. max_transmit_sizes: HashMap, + /// Cap on `message_ids` decoded per IWANT control message. + max_iwant_length: usize, + /// Cap on `message_ids` decoded per IDONTWANT control message. + max_idontwant_messages: usize, } impl GossipsubCodec { @@ -185,12 +202,16 @@ impl GossipsubCodec { max_length: usize, validation_mode: ValidationMode, max_transmit_sizes: HashMap, + max_iwant_length: usize, + max_idontwant_messages: usize, ) -> GossipsubCodec { let codec = quick_protobuf_codec::Codec::new(max_length); GossipsubCodec { validation_mode, codec, max_transmit_sizes, + max_iwant_length, + max_idontwant_messages, } } @@ -499,6 +520,7 @@ impl Decoder for GossipsubCodec { message_ids: iwant .message_ids .into_iter() + .take(self.max_iwant_length) .map(MessageId::from) .collect::>(), }) @@ -550,6 +572,7 @@ impl Decoder for GossipsubCodec { message_ids: idontwant .message_ids .into_iter() + .take(self.max_idontwant_messages) .map(MessageId::from) .collect::>(), }) @@ -662,8 +685,13 @@ mod tests { message_id: MessageId(vec![0, 0]), }; - let mut codec = - GossipsubCodec::new(u32::MAX as usize, ValidationMode::Strict, HashMap::new()); + let mut codec = GossipsubCodec::new( + u32::MAX as usize, + ValidationMode::Strict, + HashMap::new(), + 5000, + 1000, + ); let mut buf = BytesMut::new(); codec.encode(rpc.into_protobuf(), &mut buf).unwrap(); let decoded_rpc = codec.decode(&mut buf).unwrap().unwrap(); From 7bd4109bcec4b9dc6cf5afa68c0d26f1e90cef10 Mon Sep 17 00:00:00 2001 From: Adesh Kolte Date: Tue, 28 Apr 2026 12:13:19 +0530 Subject: [PATCH 2/3] docs(gossipsub): add changelog entry for IWANT/IDONTWANT decode-time cap --- protocols/gossipsub/CHANGELOG.md | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/protocols/gossipsub/CHANGELOG.md b/protocols/gossipsub/CHANGELOG.md index 61289e15b82..388720ed950 100644 --- a/protocols/gossipsub/CHANGELOG.md +++ b/protocols/gossipsub/CHANGELOG.md @@ -1,5 +1,11 @@ ## 0.50.0 +- Bound IWANT and IDONTWANT `message_ids` at decode time so the codec + applies `max_iwant_length` / `max_idontwant_messages` before allocating + the full `Vec`. The behaviour-layer `truncate` now acts as a + defense-in-depth safety net rather than the only cap. + See [PR 578](https://github.com/sigp/rust-libp2p/pull/578) + - Rename metric `topic_msg_sent_bytes` to `topic_msg_last_sent_bytes` for accuracy. See [PR 6283](https://github.com/libp2p/rust-libp2p/pull/6283) From c12f4f5d5eba3d7fe1eb266f58eb963a1989ad72 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Oliveira?= Date: Wed, 29 Apr 2026 17:08:11 +0100 Subject: [PATCH 3/3] define a max_control_messages limit for all and each control message, instead of having multiple ihave/iwant/idontwant limits --- protocols/gossipsub/CHANGELOG.md | 9 +- protocols/gossipsub/src/behaviour.rs | 45 ++-- .../gossipsub/src/behaviour/tests/gossip.rs | 10 +- .../src/behaviour/tests/topic_config.rs | 4 +- protocols/gossipsub/src/config.rs | 97 +++------ protocols/gossipsub/src/protocol.rs | 202 +++++++++--------- 6 files changed, 156 insertions(+), 211 deletions(-) diff --git a/protocols/gossipsub/CHANGELOG.md b/protocols/gossipsub/CHANGELOG.md index 388720ed950..cc37d12b33a 100644 --- a/protocols/gossipsub/CHANGELOG.md +++ b/protocols/gossipsub/CHANGELOG.md @@ -1,10 +1,9 @@ ## 0.50.0 -- Bound IWANT and IDONTWANT `message_ids` at decode time so the codec - applies `max_iwant_length` / `max_idontwant_messages` before allocating - the full `Vec`. The behaviour-layer `truncate` now acts as a - defense-in-depth safety net rather than the only cap. - See [PR 578](https://github.com/sigp/rust-libp2p/pull/578) +- Unify gossipsub control-message limits under max_control_messages (replacing per-type control ID caps), + and truncate control vectors immediately after RPC decode. + rename `max_ihave_messages` to `max_ihave_messages_heartbeat`. + See PR XXXX (https://github.com/libp2p/rust-libp2p/pull/XXXX) - Rename metric `topic_msg_sent_bytes` to `topic_msg_last_sent_bytes` for accuracy. See [PR 6283](https://github.com/libp2p/rust-libp2p/pull/6283) diff --git a/protocols/gossipsub/src/behaviour.rs b/protocols/gossipsub/src/behaviour.rs index 64db151b6cf..a11da0921d8 100644 --- a/protocols/gossipsub/src/behaviour.rs +++ b/protocols/gossipsub/src/behaviour.rs @@ -1207,7 +1207,7 @@ where // IHAVE flood protection let peer_have = self.count_received_ihave.entry(*peer_id).or_insert(0); *peer_have += 1; - if *peer_have > self.config.max_ihave_messages() { + if *peer_have > self.config.max_ihave_messages_heartbeat() { tracing::debug!( peer=%peer_id, "IHAVE: peer has advertised too many times ({}) within this heartbeat \ @@ -1218,7 +1218,7 @@ where } if let Some(iasked) = self.count_sent_iwant.get(peer_id) { - if *iasked >= self.config.max_ihave_length() { + if *iasked >= self.config.max_control_messages() { tracing::debug!( peer=%peer_id, "IHAVE: peer has already advertised too many messages ({}); ignoring", @@ -1263,8 +1263,8 @@ where if !iwant_ids.is_empty() { let iasked = self.count_sent_iwant.entry(*peer_id).or_insert(0); let mut iask = iwant_ids.len(); - if *iasked + iask > self.config.max_ihave_length() { - iask = self.config.max_ihave_length().saturating_sub(*iasked); + if *iasked + iask > self.config.max_control_messages() { + iask = self.config.max_control_messages().saturating_sub(*iasked); } // Send the list of IWANT control messages @@ -2561,7 +2561,7 @@ where } // if we are emitting more than GossipSubMaxIHaveLength message_ids, truncate the list - if message_ids.len() > self.config.max_ihave_length() { + if message_ids.len() > self.config.max_control_messages() { // we do the truncation (with shuffling) per peer below tracing::debug!( "too many messages for gossip; will truncate IHAVE list ({} messages)", @@ -2595,12 +2595,12 @@ where for peer_id in to_msg_peers { let mut peer_message_ids = message_ids.clone(); - if peer_message_ids.len() > self.config.max_ihave_length() { + if peer_message_ids.len() > self.config.max_control_messages() { // We do this per peer so that we emit a different set for each peer. // we have enough redundancy in the system that this will significantly increase // the message coverage when we do truncate. - peer_message_ids.partial_shuffle(&mut rng, self.config.max_ihave_length()); - peer_message_ids.truncate(self.config.max_ihave_length()); + peer_message_ids.partial_shuffle(&mut rng, self.config.max_control_messages()); + peer_message_ids.truncate(self.config.max_control_messages()); } // send an IHAVE message @@ -3289,16 +3289,7 @@ where } // Handle messages - for (count, raw_message) in rpc.messages.into_iter().enumerate() { - // Only process the amount of messages the configuration allows. - if self - .config - .max_messages_per_rpc() - .is_some_and(|max_msg| count >= max_msg) - { - tracing::warn!("Received more messages than permitted. Ignoring further messages. Processed: {}", count); - break; - } + for raw_message in rpc.messages { self.handle_received_message(raw_message, &propagation_source); } @@ -3308,17 +3299,7 @@ where let mut ihave_msgs = vec![]; let mut graft_msgs = vec![]; let mut prune_msgs = vec![]; - for (count, control_msg) in rpc.control_msgs.into_iter().enumerate() { - // Only process the amount of messages the configuration allows. - if self - .config - .max_messages_per_rpc() - .is_some_and(|max_msg| count >= max_msg) - { - tracing::warn!("Received more control messages than permitted. Ignoring further messages. Processed: {}", count); - break; - } - + for control_msg in rpc.control_msgs { match control_msg { ControlAction::IHave(IHave { topic_hash, @@ -3326,8 +3307,7 @@ where }) => { ihave_msgs.push((topic_hash, message_ids)); } - ControlAction::IWant(IWant { mut message_ids }) => { - message_ids.truncate(self.config.max_iwant_length()); + ControlAction::IWant(IWant { message_ids }) => { self.handle_iwant(&propagation_source, message_ids) } ControlAction::Graft(Graft { topic_hash }) => graft_msgs.push(topic_hash), @@ -3336,7 +3316,7 @@ where peers, backoff, }) => prune_msgs.push((topic_hash, peers, backoff)), - ControlAction::IDontWant(IDontWant { mut message_ids }) => { + ControlAction::IDontWant(IDontWant { message_ids }) => { let Some(peer) = self.connected_peers.get_mut(&propagation_source) else { tracing::error!(peer = %propagation_source, @@ -3344,7 +3324,6 @@ where continue; }; - message_ids.truncate(self.config.max_idontwant_messages()); // Remove messages from the queue. #[allow(unused)] let removed = peer.messages.remove_data_messages(&message_ids); diff --git a/protocols/gossipsub/src/behaviour/tests/gossip.rs b/protocols/gossipsub/src/behaviour/tests/gossip.rs index fb708cc4d98..856a0d86efa 100644 --- a/protocols/gossipsub/src/behaviour/tests/gossip.rs +++ b/protocols/gossipsub/src/behaviour/tests/gossip.rs @@ -451,7 +451,7 @@ fn test_ignore_too_many_iwants_from_same_peer_for_same_message() { #[test] fn test_ignore_too_many_ihaves() { let config = ConfigBuilder::default() - .max_ihave_messages(10) + .max_ihave_messages_heartbeat(10) .build() .unwrap(); // build gossipsub with full mesh @@ -528,8 +528,8 @@ fn test_ignore_too_many_ihaves() { #[test] fn test_ignore_too_many_messages_in_ihave() { let config = ConfigBuilder::default() - .max_ihave_messages(10) - .max_ihave_length(10) + .max_ihave_messages_heartbeat(10) + .max_control_messages(10) .build() .unwrap(); // build gossipsub with full mesh @@ -609,8 +609,8 @@ fn test_ignore_too_many_messages_in_ihave() { #[test] fn test_limit_number_of_message_ids_inside_ihave() { let config = ConfigBuilder::default() - .max_ihave_messages(10) - .max_ihave_length(100) + .max_ihave_messages_heartbeat(10) + .max_control_messages(100) .build() .unwrap(); // build gossipsub with full mesh diff --git a/protocols/gossipsub/src/behaviour/tests/topic_config.rs b/protocols/gossipsub/src/behaviour/tests/topic_config.rs index 7d5ae4bc772..4e95ef36d98 100644 --- a/protocols/gossipsub/src/behaviour/tests/topic_config.rs +++ b/protocols/gossipsub/src/behaviour/tests/topic_config.rs @@ -668,7 +668,7 @@ fn test_validation_error_message_size_too_large_topic_specific() { ValidationMode::None, max_transmit_size_map, 5000, - 1000, + 5000, ); let mut buf = BytesMut::new(); let rpc = proto::RPC { @@ -774,7 +774,7 @@ fn test_validation_message_size_within_topic_specific() { ValidationMode::None, max_transmit_size_map, 5000, - 1000, + 5000, ); let mut buf = BytesMut::new(); let rpc = proto::RPC { diff --git a/protocols/gossipsub/src/config.rs b/protocols/gossipsub/src/config.rs index 9450f1a107f..dfc3f23efab 100644 --- a/protocols/gossipsub/src/config.rs +++ b/protocols/gossipsub/src/config.rs @@ -123,11 +123,9 @@ pub struct Config { opportunistic_graft_ticks: u64, opportunistic_graft_peers: usize, gossip_retransimission: u32, - max_messages_per_rpc: Option, - max_ihave_length: usize, - max_iwant_length: usize, - max_ihave_messages: usize, - max_idontwant_messages: usize, + max_publish_messages: usize, + max_control_messages: usize, + max_ihave_messages_heartbeat: usize, iwant_followup_time: Duration, connection_handler_queue_len: usize, connection_handler_publish_duration: Duration, @@ -412,30 +410,16 @@ impl Config { self.opportunistic_graft_peers } - /// The maximum number of messages we will process in a given RPC. If this is unset, there is + /// The maximum number of publish messages we will process in a given RPC. If this is unset, there is /// no limit. The default is None. - pub fn max_messages_per_rpc(&self) -> Option { - self.max_messages_per_rpc + pub fn max_publish_messages(&self) -> usize { + self.max_publish_messages } - /// The maximum number of messages to include in an IHAVE message. - /// Also controls the maximum number of IHAVE ids we will accept and request with IWANT from a - /// peer within a heartbeat, to protect from IHAVE floods. You should adjust this value from the - /// default if your system is pushing more than 5000 messages in GossipSubHistoryGossip - /// heartbeats; with the defaults this is 1666 messages/s. The default is 5000. - pub fn max_ihave_length(&self) -> usize { - self.max_ihave_length - } - - /// The maximum number of `message_ids` that we accept in a single IWANT message. - pub fn max_iwant_length(&self) -> usize { - self.max_iwant_length - } - - /// GossipSubMaxIHaveMessages is the maximum number of IHAVE messages to accept from a peer - /// within a heartbeat. - pub fn max_ihave_messages(&self) -> usize { - self.max_ihave_messages + /// The maximum number of control messages per type we will process in a given RPC. If this is unset, there is + /// no limit. The default is None. + pub fn max_control_messages(&self) -> usize { + self.max_control_messages } /// Time to wait for a message requested through IWANT following an IHAVE advertisement. @@ -484,9 +468,10 @@ impl Config { self.idontwant_on_publish } - /// The maximum number of `message_ids` that we accept in a single IDONTWANT message. - pub fn max_idontwant_messages(&self) -> usize { - self.max_idontwant_messages + /// GossipSubMaxIHaveMessages is the maximum number of IHAVE messages to accept from a peer + /// within a heartbeat. + pub fn max_ihave_messages_heartbeat(&self) -> usize { + self.max_ihave_messages_heartbeat } } @@ -548,16 +533,14 @@ impl Default for ConfigBuilder { opportunistic_graft_ticks: 60, opportunistic_graft_peers: 2, gossip_retransimission: 3, - max_messages_per_rpc: None, - max_ihave_length: 5000, - max_iwant_length: 5000, - max_ihave_messages: 10, + max_publish_messages: 5000, + max_control_messages: 5000, + max_ihave_messages_heartbeat: 10, iwant_followup_time: Duration::from_secs(3), connection_handler_queue_len: 5000, connection_handler_publish_duration: Duration::from_secs(5), connection_handler_forward_duration: Duration::from_secs(1), idontwant_message_size_threshold: 1000, - max_idontwant_messages: 1000, idontwant_on_publish: false, topic_configuration: TopicConfigs::default(), }, @@ -969,27 +952,17 @@ impl ConfigBuilder { self } - /// The maximum number of messages we will process in a given RPC. If this is unset, there is - /// no limit. The default is None. - pub fn max_messages_per_rpc(&mut self, max: Option) -> &mut Self { - self.config.max_messages_per_rpc = max; - self - } - - /// The maximum number of messages to include in an IHAVE message. - /// Also controls the maximum number of IHAVE ids we will accept and request with IWANT from a - /// peer within a heartbeat, to protect from IHAVE floods. You should adjust this value from the - /// default if your system is pushing more than 5000 messages in GossipSubHistoryGossip - /// heartbeats; with the defaults this is 1666 messages/s. The default is 5000. - pub fn max_ihave_length(&mut self, max_ihave_length: usize) -> &mut Self { - self.config.max_ihave_length = max_ihave_length; + /// The maximum number of messages we will process in a given RPC. The default is 5000. + pub fn max_messages_per_rpc(&mut self, max: usize) -> &mut Self { + self.config.max_publish_messages = max; + self.config.protocol.max_publish_messages = max; self } /// GossipSubMaxIHaveMessages is the maximum number of IHAVE messages to accept from a peer /// within a heartbeat. - pub fn max_ihave_messages(&mut self, max_ihave_messages: usize) -> &mut Self { - self.config.max_ihave_messages = max_ihave_messages; + pub fn max_ihave_messages_heartbeat(&mut self, max_ihave_messages: usize) -> &mut Self { + self.config.max_ihave_messages_heartbeat = max_ihave_messages; self } @@ -1063,17 +1036,10 @@ impl ConfigBuilder { self } - /// The maximum number of `message_ids` that we accept in a single IDONTWANT message. - pub fn max_idontwant_messages(&mut self, size: usize) -> &mut Self { - self.config.max_idontwant_messages = size; - self.config.protocol.max_idontwant_messages = size; - self - } - - /// The maximum number of `message_ids` that we accept in a single IWANT message. - pub fn max_iwant_messages(&mut self, size: usize) -> &mut Self { - self.config.max_iwant_length = size; - self.config.protocol.max_iwant_length = size; + /// The max number of control messages to decode in a single RPC. The default is 5000. + pub fn max_control_messages(&mut self, size: usize) -> &mut Self { + self.config.max_control_messages = size; + self.config.protocol.max_control_messages = size; self } @@ -1185,9 +1151,12 @@ impl std::fmt::Debug for Config { ); let _ = builder.field("opportunistic_graft_ticks", &self.opportunistic_graft_ticks); let _ = builder.field("opportunistic_graft_peers", &self.opportunistic_graft_peers); - let _ = builder.field("max_messages_per_rpc", &self.max_messages_per_rpc); - let _ = builder.field("max_ihave_length", &self.max_ihave_length); - let _ = builder.field("max_ihave_messages", &self.max_ihave_messages); + let _ = builder.field("max_messages_per_rpc", &self.max_publish_messages); + let _ = builder.field("max_control_messages", &self.max_control_messages); + let _ = builder.field( + "max_ihave_messages_heartbeat", + &self.max_ihave_messages_heartbeat, + ); let _ = builder.field("iwant_followup_time", &self.iwant_followup_time); let _ = builder.field( "idontwant_message_size_threshold", diff --git a/protocols/gossipsub/src/protocol.rs b/protocols/gossipsub/src/protocol.rs index 70dd2cd7039..cb53ad87dae 100644 --- a/protocols/gossipsub/src/protocol.rs +++ b/protocols/gossipsub/src/protocol.rs @@ -72,13 +72,10 @@ pub struct ProtocolConfig { pub(crate) default_max_transmit_size: usize, /// The max transmit sizes for a topic. pub(crate) max_transmit_sizes: HashMap, - /// Cap on `message_ids` decoded per IWANT control message. Mirrored from - /// `Config::max_iwant_length` so the codec can bound allocations before - /// the full RPC reaches the behaviour layer. - pub(crate) max_iwant_length: usize, - /// Cap on `message_ids` decoded per IDONTWANT control message. Mirrored - /// from `Config::max_idontwant_messages`. - pub(crate) max_idontwant_messages: usize, + /// The max number of publish messages to decode in a single RPC. + pub(crate) max_publish_messages: usize, + /// The max number of control messages per type to decode in a single RPC. + pub(crate) max_control_messages: usize, } impl Default for ProtocolConfig { @@ -92,8 +89,8 @@ impl Default for ProtocolConfig { ], default_max_transmit_size: 65536, max_transmit_sizes: HashMap::new(), - max_iwant_length: 5000, - max_idontwant_messages: 1000, + max_publish_messages: 500, + max_control_messages: 500, } } } @@ -148,8 +145,8 @@ where self.default_max_transmit_size, self.validation_mode, self.max_transmit_sizes, - self.max_iwant_length, - self.max_idontwant_messages, + self.max_publish_messages, + self.max_control_messages, ), ), protocol_id.kind, @@ -173,8 +170,8 @@ where self.default_max_transmit_size, self.validation_mode, self.max_transmit_sizes, - self.max_iwant_length, - self.max_idontwant_messages, + self.max_publish_messages, + self.max_control_messages, ), ), protocol_id.kind, @@ -191,10 +188,10 @@ pub struct GossipsubCodec { codec: quick_protobuf_codec::Codec, /// Maximum transmit sizes per topic, with a default if not specified. max_transmit_sizes: HashMap, - /// Cap on `message_ids` decoded per IWANT control message. - max_iwant_length: usize, - /// Cap on `message_ids` decoded per IDONTWANT control message. - max_idontwant_messages: usize, + /// The max number of publish messages to decode in a single RPC. + max_publish_messages: usize, + /// The max number of control messages per type to decode in a single RPC. + max_control_messages: usize, } impl GossipsubCodec { @@ -202,16 +199,16 @@ impl GossipsubCodec { max_length: usize, validation_mode: ValidationMode, max_transmit_sizes: HashMap, - max_iwant_length: usize, - max_idontwant_messages: usize, + max_publish_messages: usize, + max_control_messages: usize, ) -> GossipsubCodec { let codec = quick_protobuf_codec::Codec::new(max_length); GossipsubCodec { validation_mode, codec, max_transmit_sizes, - max_iwant_length, - max_idontwant_messages, + max_publish_messages, + max_control_messages, } } @@ -291,9 +288,19 @@ impl Decoder for GossipsubCodec { type Error = quick_protobuf_codec::Error; fn decode(&mut self, src: &mut BytesMut) -> Result, Self::Error> { - let Some(rpc) = self.codec.decode(src)? else { + let Some(mut rpc) = self.codec.decode(src)? else { return Ok(None); }; + + // Limit messages + rpc.publish.truncate(self.max_publish_messages); + let mut control = rpc.control.take().unwrap_or_default(); + control.ihave.truncate(self.max_control_messages); + control.iwant.truncate(self.max_control_messages); + control.graft.truncate(self.max_control_messages); + control.prune.truncate(self.max_control_messages); + control.idontwant.truncate(self.max_control_messages); + // Store valid messages. let mut messages = Vec::with_capacity(rpc.publish.len()); // Store any invalid messages. @@ -495,96 +502,87 @@ impl Decoder for GossipsubCodec { let mut control_msgs = Vec::new(); - if let Some(rpc_control) = rpc.control { - // Collect the gossipsub control messages - let ihave_msgs: Vec = rpc_control - .ihave - .into_iter() - .map(|ihave| { - ControlAction::IHave(IHave { - topic_hash: TopicHash::from_raw(ihave.topic_id.unwrap_or_default()), - message_ids: ihave - .message_ids - .into_iter() - .map(MessageId::from) - .collect::>(), - }) + // Collect the gossipsub control messages + let ihave_msgs: Vec = control + .ihave + .into_iter() + .map(|ihave| { + ControlAction::IHave(IHave { + topic_hash: TopicHash::from_raw(ihave.topic_id.unwrap_or_default()), + message_ids: ihave + .message_ids + .into_iter() + .map(MessageId::from) + .collect::>(), }) - .collect(); - - let iwant_msgs: Vec = rpc_control - .iwant - .into_iter() - .map(|iwant| { - ControlAction::IWant(IWant { - message_ids: iwant - .message_ids - .into_iter() - .take(self.max_iwant_length) - .map(MessageId::from) - .collect::>(), - }) + }) + .collect(); + control_msgs.extend(ihave_msgs); + + let iwant_msgs: Vec = control + .iwant + .into_iter() + .map(|iwant| { + ControlAction::IWant(IWant { + message_ids: iwant + .message_ids + .into_iter() + .map(MessageId::from) + .collect::>(), }) - .collect(); - - let graft_msgs: Vec = rpc_control - .graft - .into_iter() - .map(|graft| { - ControlAction::Graft(Graft { - topic_hash: TopicHash::from_raw(graft.topic_id.unwrap_or_default()), - }) + }) + .collect(); + control_msgs.extend(iwant_msgs); + + let graft_msgs: Vec = control + .graft + .into_iter() + .map(|graft| { + ControlAction::Graft(Graft { + topic_hash: TopicHash::from_raw(graft.topic_id.unwrap_or_default()), }) - .collect(); - - let mut prune_msgs = Vec::new(); - - for prune in rpc_control.prune { - // filter out invalid peers - let peers = prune - .peers - .into_iter() - .filter_map(|info| { - info.peer_id - .as_ref() - .and_then(|id| PeerId::from_bytes(id).ok()) - .map(|peer_id| + }) + .collect(); + control_msgs.extend(graft_msgs); + + let mut prune_messages = Vec::new(); + for prune in control.prune { + // filter out invalid peers + let peers = prune + .peers + .into_iter() + .filter_map(|info| { + info.peer_id + .as_ref() + .and_then(|id| PeerId::from_bytes(id).ok()) + .map(|peer_id| //TODO signedPeerRecord, see https://github.com/libp2p/specs/pull/217 PeerInfo { peer_id: Some(peer_id), }) - }) - .collect::>(); - - let topic_hash = TopicHash::from_raw(prune.topic_id.unwrap_or_default()); - prune_msgs.push(ControlAction::Prune(Prune { - topic_hash, - peers, - backoff: prune.backoff, - })); - } - - let idontwant_msgs: Vec = rpc_control - .idontwant - .into_iter() - .map(|idontwant| { - ControlAction::IDontWant(IDontWant { - message_ids: idontwant - .message_ids - .into_iter() - .take(self.max_idontwant_messages) - .map(MessageId::from) - .collect::>(), - }) }) - .collect(); + .collect::>(); + + let topic_hash = TopicHash::from_raw(prune.topic_id.unwrap_or_default()); + prune_messages.push(ControlAction::Prune(Prune { + topic_hash, + peers, + backoff: prune.backoff, + })); + } + control_msgs.extend(prune_messages); - control_msgs.extend(ihave_msgs); - control_msgs.extend(iwant_msgs); - control_msgs.extend(graft_msgs); - control_msgs.extend(prune_msgs); - control_msgs.extend(idontwant_msgs); + let mut idontwant_messages = Vec::new(); + for idontwant in control.idontwant { + idontwant_messages.push(ControlAction::IDontWant(IDontWant { + message_ids: idontwant + .message_ids + .into_iter() + .map(MessageId::from) + .collect::>(), + })); } + control_msgs.extend(idontwant_messages); Ok(Some(HandlerEvent::Message { rpc: RpcIn {