libp2p_gossipsub/
gossip_promises.rs

1// Copyright 2020 Sigma Prime Pty Ltd.
2//
3// Permission is hereby granted, free of charge, to any person obtaining a
4// copy of this software and associated documentation files (the "Software"),
5// to deal in the Software without restriction, including without limitation
6// the rights to use, copy, modify, merge, publish, distribute, sublicense,
7// and/or sell copies of the Software, and to permit persons to whom the
8// Software is furnished to do so, subject to the following conditions:
9//
10// The above copyright notice and this permission notice shall be included in
11// all copies or substantial portions of the Software.
12//
13// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
14// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
15// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
16// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
17// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
18// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
19// DEALINGS IN THE SOFTWARE.
20
21use std::collections::HashMap;
22
23use libp2p_identity::PeerId;
24use web_time::Instant;
25
26use crate::{peer_score::RejectReason, MessageId, ValidationError};
27
28/// Tracks recently sent `IWANT` messages and checks if peers respond to them.
29#[derive(Default)]
30pub(crate) struct GossipPromises {
31    /// Stores for each tracked message id and peer the instant when this promise expires.
32    ///
33    /// If the peer didn't respond until then we consider the promise as broken and penalize the
34    /// peer.
35    promises: HashMap<MessageId, HashMap<PeerId, Instant>>,
36}
37
38impl GossipPromises {
39    /// Returns true if the message id exists in the promises.
40    pub(crate) fn contains(&self, message: &MessageId) -> bool {
41        self.promises.contains_key(message)
42    }
43
44    /// Get the peers we sent IWANT the input message id.
45    pub(crate) fn peers_for_message(&self, message_id: &MessageId) -> Vec<PeerId> {
46        self.promises
47            .get(message_id)
48            .map(|peers| peers.keys().copied().collect())
49            .unwrap_or_default()
50    }
51
52    /// Track a promise to deliver a message from a list of [`MessageId`]s we are requesting.
53    pub(crate) fn add_promise(&mut self, peer: PeerId, messages: &[MessageId], expires: Instant) {
54        for message_id in messages {
55            // If a promise for this message id and peer already exists we don't update the expiry!
56            self.promises
57                .entry(message_id.clone())
58                .or_default()
59                .entry(peer)
60                .or_insert(expires);
61        }
62    }
63
64    pub(crate) fn message_delivered(&mut self, message_id: &MessageId) {
65        // Someone delivered a message, we can stop tracking all promises for it.
66        self.promises.remove(message_id);
67    }
68
69    pub(crate) fn reject_message(&mut self, message_id: &MessageId, reason: &RejectReason) {
70        // A message got rejected, so we can stop tracking promises and let the score penalty apply
71        // from invalid message delivery.
72        // We do take exception and apply promise penalty regardless in the following cases, where
73        // the peer delivered an obviously invalid message.
74        match reason {
75            RejectReason::ValidationError(ValidationError::InvalidSignature) => (),
76            RejectReason::SelfOrigin => (),
77            _ => {
78                self.promises.remove(message_id);
79            }
80        };
81    }
82
83    /// Returns the number of broken promises for each peer who didn't follow up on an IWANT
84    /// request.
85    /// This should be called not too often relative to the expire times, since it iterates over
86    /// the whole stored data.
87    pub(crate) fn get_broken_promises(&mut self) -> HashMap<PeerId, usize> {
88        let now = Instant::now();
89        let mut result = HashMap::new();
90        self.promises.retain(|msg, peers| {
91            peers.retain(|peer_id, expires| {
92                if *expires < now {
93                    let count = result.entry(*peer_id).or_insert(0);
94                    *count += 1;
95                    tracing::debug!(
96                        peer=%peer_id,
97                        message=%msg,
98                        "[Penalty] The peer broke the promise to deliver message in time!"
99                    );
100                    false
101                } else {
102                    true
103                }
104            });
105            !peers.is_empty()
106        });
107        result
108    }
109}
libp2p_gossipsub/gossip_promises.rs

libp2p_gossipsub/
gossip_promises.rs
