referrerpolicy=no-referrer-when-downgrade
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118

// Copyright (C) Parity Technologies (UK) Ltd.
// This file is part of Polkadot.

// Polkadot is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.

// Polkadot is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.

// You should have received a copy of the GNU General Public License
// along with Polkadot.  If not, see <http://www.gnu.org/licenses/>.

//! A utility abstraction to collect and send reputation changes.

use polkadot_node_network_protocol::{PeerId, UnifiedReputationChange};
use polkadot_node_subsystem::{
	messages::{NetworkBridgeTxMessage, ReportPeerMessage},
	overseer,
};
use std::{collections::HashMap, time::Duration};

/// Default delay for sending reputation changes
pub const REPUTATION_CHANGE_INTERVAL: Duration = Duration::from_secs(30);
const LOG_TARGET: &'static str = "parachain::reputation-aggregator";

type BatchReputationChange = HashMap<PeerId, i32>;

/// Collects reputation changes and sends them in one batch to relieve network channels
#[derive(Debug, Clone)]
pub struct ReputationAggregator {
	send_immediately_if: fn(UnifiedReputationChange) -> bool,
	by_peer: Option<BatchReputationChange>,
}

impl Default for ReputationAggregator {
	fn default() -> Self {
		Self::new(|rep| matches!(rep, UnifiedReputationChange::Malicious(_)))
	}
}

impl ReputationAggregator {
	/// New `ReputationAggregator`
	///
	/// # Arguments
	///
	/// * `send_immediately_if` - A function, takes `UnifiedReputationChange`,
	/// results shows if we need to send the changes right away.
	/// By default, it is used for sending `UnifiedReputationChange::Malicious` changes immediately
	/// and for testing.
	pub fn new(send_immediately_if: fn(UnifiedReputationChange) -> bool) -> Self {
		Self { by_peer: Default::default(), send_immediately_if }
	}

	/// Sends collected reputation changes in a batch,
	/// removing them from inner state
	pub async fn send(
		&mut self,
		sender: &mut impl overseer::SubsystemSender<NetworkBridgeTxMessage>,
	) {
		if let Some(by_peer) = self.by_peer.take() {
			sender
				.send_message(NetworkBridgeTxMessage::ReportPeer(ReportPeerMessage::Batch(by_peer)))
				.await;
		}
	}

	/// Adds reputation change to inner state
	/// or sends it right away if the change is dangerous
	pub async fn modify(
		&mut self,
		sender: &mut impl overseer::SubsystemSender<NetworkBridgeTxMessage>,
		peer_id: PeerId,
		rep: UnifiedReputationChange,
	) {
		if rep.cost_or_benefit() < 0 {
			gum::debug!(target: LOG_TARGET, peer = ?peer_id, ?rep, "Reduce reputation");
		}

		if (self.send_immediately_if)(rep) {
			self.single_send(sender, peer_id, rep).await;
		} else {
			self.add(peer_id, rep);
		}
	}

	async fn single_send(
		&self,
		sender: &mut impl overseer::SubsystemSender<NetworkBridgeTxMessage>,
		peer_id: PeerId,
		rep: UnifiedReputationChange,
	) {
		sender
			.send_message(NetworkBridgeTxMessage::ReportPeer(ReportPeerMessage::Single(
				peer_id,
				rep.into(),
			)))
			.await;
	}

	fn add(&mut self, peer_id: PeerId, rep: UnifiedReputationChange) {
		let by_peer = self.by_peer.get_or_insert(HashMap::new());
		add_reputation(by_peer, peer_id, rep)
	}
}

/// Add a reputation change to an existing collection.
pub fn add_reputation(
	acc: &mut BatchReputationChange,
	peer_id: PeerId,
	rep: UnifiedReputationChange,
) {
	let cost = rep.cost_or_benefit();
	acc.entry(peer_id).and_modify(|v| *v = v.saturating_add(cost)).or_insert(cost);
}