lib.rs 4.22 KB
Newer Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
// Copyright 2020 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/>.

//! The Collator Protocol allows collators and validators talk to each other.
//! This subsystem implements both sides of the collator protocol.

20
#![deny(missing_docs, unused_crate_dependencies)]
21
22

use std::time::Duration;
23
use futures::{channel::oneshot, FutureExt, TryFutureExt};
24
use log::trace;
25
use thiserror::Error;
26
27
28
29
30
31
32
33
34
35
36
37

use polkadot_subsystem::{
	Subsystem, SubsystemContext, SubsystemError, SpawnedSubsystem,
	errors::RuntimeApiError,
	messages::{
		AllMessages, CollatorProtocolMessage, NetworkBridgeMessage,
	},
};
use polkadot_node_network_protocol::{
	PeerId, ReputationChange as Rep,
};
use polkadot_primitives::v1::CollatorId;
38
39
use polkadot_node_subsystem_util::{
	self as util,
40
	metrics::prometheus,
41
};
42
43
44
45
46
47
48

mod collator_side;
mod validator_side;

const TARGET: &'static str = "colp";
const REQUEST_TIMEOUT: Duration = Duration::from_secs(1);

49
#[derive(Debug, Error)]
50
enum Error {
51
52
53
54
55
56
57
58
59
60
	#[error(transparent)]
	Subsystem(#[from] SubsystemError),
	#[error(transparent)]
	Oneshot(#[from] oneshot::Canceled),
	#[error(transparent)]
	RuntimeApi(#[from] RuntimeApiError),
	#[error(transparent)]
	UtilError(#[from] util::Error),
	#[error(transparent)]
	Prometheus(#[from] prometheus::PrometheusError),
61
62
}

63
64
65
66
67
68
69
70
71
72
impl From<util::validator_discovery::Error> for Error {
	fn from(me: util::validator_discovery::Error) -> Self {
		match me {
			util::validator_discovery::Error::Subsystem(s) => Error::Subsystem(s),
			util::validator_discovery::Error::RuntimeApi(ra) => Error::RuntimeApi(ra),
			util::validator_discovery::Error::Oneshot(c) => Error::Oneshot(c),
		}
	}
}

73
74
type Result<T> = std::result::Result<T, Error>;

75
76
77
/// What side of the collator protocol is being engaged
pub enum ProtocolSide {
	/// Validators operate on the relay chain.
78
	Validator(validator_side::Metrics),
79
	/// Collators operate on a parachain.
80
	Collator(CollatorId, collator_side::Metrics),
81
82
83
84
}

/// The collator protocol subsystem.
pub struct CollatorProtocolSubsystem {
85
	protocol_side: ProtocolSide,
86
87
88
89
90
}

impl CollatorProtocolSubsystem {
	/// Start the collator protocol.
	/// If `id` is `Some` this is a collator side of the protocol.
91
	/// If `id` is `None` this is a validator side of the protocol.
92
	/// Caller must provide a registry for prometheus metrics.
93
	pub fn new(protocol_side: ProtocolSide) -> Self {
94
95
96
97
98
99
100
101
102
103
		Self {
			protocol_side,
		}
	}

	async fn run<Context>(self, ctx: Context) -> Result<()>
	where
		Context: SubsystemContext<Message = CollatorProtocolMessage>,
	{
		match self.protocol_side {
104
105
106
107
108
109
110
111
112
113
		    ProtocolSide::Validator(metrics) => validator_side::run(
				ctx,
				REQUEST_TIMEOUT,
				metrics,
			).await,
		    ProtocolSide::Collator(id, metrics) => collator_side::run(
				ctx,
				id,
				metrics,
			).await,
114
115
116
		}.map_err(|e| {
			SubsystemError::with_origin("collator-protocol", e).into()
		})
117
118
119
120
121
122
123
124
	}
}

impl<Context> Subsystem<Context> for CollatorProtocolSubsystem
where
	Context: SubsystemContext<Message = CollatorProtocolMessage> + Sync + Send,
{
	fn start(self, ctx: Context) -> SpawnedSubsystem {
125
126
127
128
129
		let future = self
			.run(ctx)
			.map_err(|e| SubsystemError::with_origin("collator-protocol", e))
			.boxed();

130
131
		SpawnedSubsystem {
			name: "collator-protocol-subsystem",
132
			future,
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
		}
	}
}

/// Modify the reputation of a peer based on its behavior.
async fn modify_reputation<Context>(ctx: &mut Context, peer: PeerId, rep: Rep) -> Result<()>
where
	Context: SubsystemContext<Message = CollatorProtocolMessage>,
{
	trace!(
		target: TARGET,
		"Reputation change of {:?} for peer {:?}", rep, peer,
	);

	ctx.send_message(AllMessages::NetworkBridge(
		NetworkBridgeMessage::ReportPeer(peer, rep),
	)).await?;

	Ok(())
}