Skip to content

Commit 5f7eb90

Browse files
dknopikdknopik
authored
fix(gossipsub): Count forwarded messages in metrics
Forwarded messages were not accounted in the `topic_msg_sent_*` metrics. Furthermore, after this change, all code location matching on `RpcOut` handle `Publish` and `Forward` the same, so I unified them into `Publish`. Pull-Request: libp2p#6502.
1 parent a714395 commit 5f7eb90

9 files changed

Lines changed: 25 additions & 48 deletions

File tree

‎protocols/gossipsub/CHANGELOG.md‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,7 @@
11
## 0.50.0
2+
- Account for forwarded messages in `topic_mesg_sent_*` metrics.
3+
See [PR 6502](https://github.com/libp2p/rust-libp2p/pull/6502)
4+
25
- Simplify gossipsub control message validation by scanning protobuf bytes before decoding
36
to validate cumulative control message size, rather than decoding then counting individual IDs
47
to prevent memory amplification from decoding.

‎protocols/gossipsub/src/behaviour.rs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1501,7 +1501,7 @@ where
15011501
tracing::debug!(peer=%peer_id, "IWANT: Sending cached messages to peer");
15021502
self.send_message(
15031503
*peer_id,
1504-
RpcOut::Forward {
1504+
RpcOut::Publish {
15051505
message_id: id.clone(),
15061506
message: msg,
15071507
timeout: Delay::new(self.config.forward_queue_duration()),
@@ -3025,7 +3025,7 @@ where
30253025

30263026
self.send_message(
30273027
*peer_id,
3028-
RpcOut::Forward {
3028+
RpcOut::Publish {
30293029
message_id: msg_id.clone(),
30303030
message: message.clone(),
30313031
timeout: Delay::new(self.config.forward_queue_duration()),

‎protocols/gossipsub/src/behaviour/tests/explicit_peers.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -235,7 +235,7 @@ fn do_forward_messages_to_explicit_peers() {
235235
assert_eq!(
236236
queues.into_iter().fold(0, |mut fwds, (peer_id, mut queue)| {
237237
while !queue.is_empty() {
238-
if matches!(queue.try_pop(), Some(RpcOut::Forward{message: m, ..}) if peer_id == peers[0] && m.data == message.data) {
238+
if matches!(queue.try_pop(), Some(RpcOut::Publish{message: m, ..}) if peer_id == peers[0] && m.data == message.data) {
239239
fwds +=1;
240240
}
241241
}

‎protocols/gossipsub/src/behaviour/tests/gossip.rs‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -72,7 +72,7 @@ fn test_handle_iwant_msg_cached() {
7272
.into_values()
7373
.fold(vec![], |mut collected_messages, mut queue| {
7474
while !queue.is_empty() {
75-
if let Some(RpcOut::Forward { message, .. }) = queue.try_pop() {
75+
if let Some(RpcOut::Publish { message, .. }) = queue.try_pop() {
7676
collected_messages.push(message)
7777
}
7878
}
@@ -129,7 +129,7 @@ fn test_handle_iwant_msg_cached_shifted() {
129129
.into_iter()
130130
.map(|(peer_id, mut queue)| {
131131
while !queue.is_empty() {
132-
if matches!(queue.try_pop(), Some(RpcOut::Forward{message, ..}) if
132+
if matches!(queue.try_pop(), Some(RpcOut::Publish{message, ..}) if
133133
gs.config.message_id(
134134
&gs.data_transform
135135
.inbound_transform(message.clone())
@@ -438,7 +438,7 @@ fn test_ignore_too_many_iwants_from_same_peer_for_same_message() {
438438
assert_eq!(
439439
queues.into_values().fold(0, |mut fwds, mut queue| {
440440
while !queue.is_empty() {
441-
if let Some(RpcOut::Forward { .. }) = queue.try_pop() {
441+
if let Some(RpcOut::Publish { .. }) = queue.try_pop() {
442442
fwds += 1;
443443
}
444444
}

‎protocols/gossipsub/src/behaviour/tests/idontwant.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -170,7 +170,7 @@ fn doesnt_forward_idontwant() {
170170
.into_iter()
171171
.fold(0, |mut fwds, (peer_id, mut queue)| {
172172
while !queue.is_empty() {
173-
if let Some(RpcOut::Forward { .. }) = queue.try_pop() {
173+
if let Some(RpcOut::Publish { .. }) = queue.try_pop() {
174174
assert_ne!(peer_id, peers[2]);
175175
fwds += 1;
176176
}

‎protocols/gossipsub/src/behaviour/tests/scoring.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -367,7 +367,7 @@ fn test_iwant_msg_from_peer_below_gossip_threshold_gets_ignored() {
367367
.into_iter()
368368
.fold(vec![], |mut collected_messages, (peer_id, mut queue)| {
369369
while !queue.is_empty() {
370-
if let Some(RpcOut::Forward { message, .. }) = queue.try_pop() {
370+
if let Some(RpcOut::Publish { message, .. }) = queue.try_pop() {
371371
collected_messages.push((peer_id, message));
372372
}
373373
}

‎protocols/gossipsub/src/handler.rs‎

Lines changed: 11 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -263,28 +263,17 @@ impl EnabledHandler {
263263
if let Poll::Ready(mut message) = Pin::new(&mut self.message_queue).poll_pop(cx)
264264
{
265265
tracing::debug!(peer=%self.peer_id, ?message, "Sending gossipsub message");
266-
match message {
267-
RpcOut::Publish {
268-
message: _,
269-
ref mut timeout,
270-
..
271-
}
272-
| RpcOut::Forward {
273-
message: _,
274-
ref mut timeout,
275-
..
276-
} => {
277-
#[allow(clippy::collapsible_match)]
278-
if Pin::new(timeout).poll(cx).is_ready() {
279-
// Inform the behaviour and end the poll.
280-
self.outbound_substream =
281-
Some(OutboundSubstreamState::WaitingOutput(substream));
282-
return Poll::Ready(ConnectionHandlerEvent::NotifyBehaviour(
283-
HandlerEvent::MessageDropped(message),
284-
));
285-
}
286-
}
287-
_ => {} // All other messages are not time-bound.
266+
if let RpcOut::Publish {
267+
ref mut timeout, ..
268+
} = message
269+
&& Pin::new(timeout).poll(cx).is_ready()
270+
{
271+
// Inform the behaviour and end the poll.
272+
self.outbound_substream =
273+
Some(OutboundSubstreamState::WaitingOutput(substream));
274+
return Poll::Ready(ConnectionHandlerEvent::NotifyBehaviour(
275+
HandlerEvent::MessageDropped(message),
276+
));
288277
}
289278
self.outbound_substream = Some(OutboundSubstreamState::PendingSend(
290279
substream,

‎protocols/gossipsub/src/queue.rs‎

Lines changed: 1 addition & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -75,7 +75,6 @@ impl Queue {
7575
self.control.try_push(message)
7676
}
7777
RpcOut::Publish { .. }
78-
| RpcOut::Forward { .. }
7978
| RpcOut::IHave(_)
8079
| RpcOut::TestExtension
8180
| RpcOut::IWant(_) => self.non_priority.try_push(message),
@@ -93,9 +92,7 @@ impl Queue {
9392

9493
let mut count = 0;
9594
self.non_priority.retain(|message| match message {
96-
RpcOut::Publish { message_id, .. } | RpcOut::Forward { message_id, .. }
97-
if message_ids.contains(message_id) =>
98-
{
95+
RpcOut::Publish { message_id, .. } if message_ids.contains(message_id) => {
9996
count += 1;
10097
false
10198
}

‎protocols/gossipsub/src/types.rs‎

Lines changed: 2 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -336,19 +336,13 @@ pub struct Extensions {
336336
#[derive(Debug)]
337337
pub enum RpcOut {
338338
/// Publish a Gossipsub message on network.`timeout` limits the duration the message
339-
/// can wait to be sent before it is abandoned.
339+
/// can wait to be sent before it is abandoned. This can be both a message originating
340+
/// from this node, or a forwarded message.
340341
Publish {
341342
message_id: MessageId,
342343
message: RawMessage,
343344
timeout: Delay,
344345
},
345-
/// Forward a Gossipsub message on network. `timeout` limits the duration the message
346-
/// can wait to be sent before it is abandoned.
347-
Forward {
348-
message_id: MessageId,
349-
message: RawMessage,
350-
timeout: Delay,
351-
},
352346
/// Subscribe a topic.
353347
Subscribe {
354348
topic: TopicHash,
@@ -410,12 +404,6 @@ impl From<RpcOut> for proto::Rpc {
410404
control: None,
411405
partial: None,
412406
},
413-
RpcOut::Forward { message, .. } => proto::Rpc {
414-
publish: vec![message.into()],
415-
subscriptions: Vec::new(),
416-
control: None,
417-
partial: None,
418-
},
419407
RpcOut::Subscribe {
420408
topic,
421409
requests_partial,

0 commit comments

Comments
 (0)