referrerpolicy=no-referrer-when-downgrade

polkadot_collator_protocol/
lib.rs

1// Copyright (C) Parity Technologies (UK) Ltd.
2// This file is part of Polkadot.
3
4// Polkadot is free software: you can redistribute it and/or modify
5// it under the terms of the GNU General Public License as published by
6// the Free Software Foundation, either version 3 of the License, or
7// (at your option) any later version.
8
9// Polkadot is distributed in the hope that it will be useful,
10// but WITHOUT ANY WARRANTY; without even the implied warranty of
11// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
12// GNU General Public License for more details.
13
14// You should have received a copy of the GNU General Public License
15// along with Polkadot.  If not, see <http://www.gnu.org/licenses/>.
16
17//! The Collator Protocol allows collators and validators talk to each other.
18//! This subsystem implements both sides of the collator protocol.
19
20#![deny(missing_docs)]
21#![deny(unused_crate_dependencies)]
22#![recursion_limit = "256"]
23
24use std::time::{Duration, Instant};
25
26use futures::{
27	stream::{FusedStream, StreamExt},
28	FutureExt, TryFutureExt,
29};
30
31use polkadot_node_subsystem_util::reputation::ReputationAggregator;
32use sp_keystore::KeystorePtr;
33
34use polkadot_node_network_protocol::{
35	request_response::{v2 as protocol_v2, IncomingRequestReceiver},
36	PeerId, UnifiedReputationChange as Rep,
37};
38use polkadot_primitives::CollatorPair;
39
40use polkadot_node_subsystem::{errors::SubsystemError, overseer, DummySubsystem, SpawnedSubsystem};
41
42mod collator_side;
43mod validator_side;
44#[cfg(feature = "experimental-collator-protocol")]
45mod validator_side_experimental;
46
47const LOG_TARGET: &'static str = "parachain::collator-protocol";
48const LOG_TARGET_STATS: &'static str = "parachain::collator-protocol-stats";
49
50/// A collator eviction policy - how fast to evict collators which are inactive.
51#[derive(Debug, Clone, Copy)]
52pub struct CollatorEvictionPolicy {
53	/// How fast to evict collators who are inactive.
54	pub inactive_collator: Duration,
55	/// How fast to evict peers which don't declare their para.
56	pub undeclared: Duration,
57}
58
59impl Default for CollatorEvictionPolicy {
60	fn default() -> Self {
61		CollatorEvictionPolicy {
62			inactive_collator: Duration::from_secs(24),
63			undeclared: Duration::from_secs(1),
64		}
65	}
66}
67
68/// What side of the collator protocol is being engaged
69pub enum ProtocolSide {
70	/// Validators operate on the relay chain.
71	Validator {
72		/// The keystore holding validator keys.
73		keystore: KeystorePtr,
74		/// An eviction policy for inactive peers or validators.
75		eviction_policy: CollatorEvictionPolicy,
76		/// Prometheus metrics for validators.
77		metrics: validator_side::Metrics,
78	},
79	/// Experimental variant of the validator side. Do not use in production.
80	#[cfg(feature = "experimental-collator-protocol")]
81	ValidatorExperimental {
82		/// The keystore holding validator keys.
83		keystore: KeystorePtr,
84		/// Prometheus metrics for validators.
85		metrics: validator_side_experimental::Metrics,
86	},
87	/// Collators operate on a parachain.
88	Collator {
89		/// Local peer id.
90		peer_id: PeerId,
91		/// Parachain collator pair.
92		collator_pair: CollatorPair,
93		/// Receiver for v2 collation fetching requests.
94		request_receiver_v2: IncomingRequestReceiver<protocol_v2::CollationFetchingRequest>,
95		/// Metrics.
96		metrics: collator_side::Metrics,
97	},
98	/// No protocol side, just disable it.
99	None,
100}
101
102/// The collator protocol subsystem.
103pub struct CollatorProtocolSubsystem {
104	protocol_side: ProtocolSide,
105}
106
107#[overseer::contextbounds(CollatorProtocol, prefix = self::overseer)]
108impl CollatorProtocolSubsystem {
109	/// Start the collator protocol.
110	/// If `id` is `Some` this is a collator side of the protocol.
111	/// If `id` is `None` this is a validator side of the protocol.
112	/// Caller must provide a registry for prometheus metrics.
113	pub fn new(protocol_side: ProtocolSide) -> Self {
114		Self { protocol_side }
115	}
116}
117
118#[overseer::subsystem(CollatorProtocol, error=SubsystemError, prefix=self::overseer)]
119impl<Context> CollatorProtocolSubsystem {
120	fn start(self, ctx: Context) -> SpawnedSubsystem {
121		let future = match self.protocol_side {
122			ProtocolSide::Validator { keystore, eviction_policy, metrics } =>
123				validator_side::run(ctx, keystore, eviction_policy, metrics)
124					.map_err(|e| SubsystemError::with_origin("collator-protocol", e))
125					.boxed(),
126			#[cfg(feature = "experimental-collator-protocol")]
127			ProtocolSide::ValidatorExperimental { keystore, metrics } =>
128				validator_side_experimental::run(ctx, keystore, metrics)
129					.map_err(|e| SubsystemError::with_origin("collator-protocol", e))
130					.boxed(),
131			ProtocolSide::Collator { peer_id, collator_pair, request_receiver_v2, metrics } =>
132				collator_side::run(ctx, peer_id, collator_pair, request_receiver_v2, metrics)
133					.map_err(|e| SubsystemError::with_origin("collator-protocol", e))
134					.boxed(),
135			ProtocolSide::None => return DummySubsystem.start(ctx),
136		};
137
138		SpawnedSubsystem { name: "collator-protocol-subsystem", future }
139	}
140}
141
142/// Modify the reputation of a peer based on its behavior.
143async fn modify_reputation(
144	reputation: &mut ReputationAggregator,
145	sender: &mut impl overseer::CollatorProtocolSenderTrait,
146	peer: PeerId,
147	rep: Rep,
148) {
149	gum::trace!(
150		target: LOG_TARGET,
151		rep = ?rep,
152		peer_id = %peer,
153		"reputation change for peer",
154	);
155
156	reputation.modify(sender, peer, rep).await;
157}
158
159/// Wait until tick and return the timestamp for the following one.
160async fn wait_until_next_tick(last_poll: Instant, period: Duration) -> Instant {
161	let now = Instant::now();
162	let next_poll = last_poll + period;
163
164	if next_poll > now {
165		futures_timer::Delay::new(next_poll - now).await
166	}
167
168	Instant::now()
169}
170
171/// Returns an infinite stream that yields with an interval of `period`.
172fn tick_stream(period: Duration) -> impl FusedStream<Item = ()> {
173	futures::stream::unfold(Instant::now(), move |next_check| async move {
174		Some(((), wait_until_next_tick(next_check, period).await))
175	})
176	.fuse()
177}