lib.rs 13.3 KB
Newer Older
Shawn Tabrizi's avatar
Shawn Tabrizi committed
1
// Copyright 2017-2020 Parity Technologies (UK) Ltd.
Gav's avatar
Gav committed
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
// 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/>.

17
//! Collation node logic.
Gav's avatar
Gav committed
18
19
20
21
22
23
24
25
26
27
28
29
30
//!
//! A collator node lives on a distinct parachain and submits a proposal for
//! a state transition, along with a proof for its validity
//! (what we might call a witness or block data).
//!
//! One of collators' other roles is to route messages between chains.
//! Each parachain produces a list of "egress" posts of messages for each other
//! parachain on each block, for a total of N^2 lists all together.
//!
//! We will refer to the egress list at relay chain block X of parachain A with
//! destination B as egress(X)[A -> B]
//!
//! On every block, each parachain will be intended to route messages from some
31
//! subset of all the other parachains. (NOTE: in practice this is not done until PoC-3)
Gav's avatar
Gav committed
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
//!
//! Since the egress information is unique to every block, when routing from a
//! parachain a collator must gather all egress posts from that parachain
//! up to the last point in history that messages were successfully routed
//! from that parachain, accounting for relay chain blocks where no candidate
//! from the collator's parachain was produced.
//!
//! In the case that all parachains route to each other and a candidate for the
//! collator's parachain was included in the last relay chain block, the collator
//! only has to gather egress posts from other parachains one block back in relay
//! chain history.
//!
//! This crate defines traits which provide context necessary for collation logic
//! to be performed, as the collation logic itself.

47
use std::collections::HashSet;
48
use std::fmt;
49
use std::sync::Arc;
50
use std::time::Duration;
51
use std::pin::Pin;
Gav's avatar
Gav committed
52

53
use futures::{future, Future, Stream, FutureExt, TryFutureExt, StreamExt, task::Spawn};
54
use log::warn;
55
use sc_client::BlockchainEvents;
Gavin Wood's avatar
Gavin Wood committed
56
57
use sp_core::Pair;
use sp_runtime::traits::BlakeTwo256;
58
use polkadot_primitives::{
59
	BlockId, Hash, Block,
60
	parachain::{
Ashley's avatar
Ashley committed
61
		self, BlockData, DutyRoster, HeadData, Id as ParaId,
62
		PoVBlock, ValidatorId, CollatorPair, LocalValidationData
63
64
65
	}
};
use polkadot_cli::{
66
	ProvideRuntimeApi, AbstractService, ParachainHost, IsKusama,
67
	service::{self, Role}
68
};
Cecile Tonglet's avatar
WIP    
Cecile Tonglet committed
69
pub use polkadot_cli::service::Configuration;
Cecile Tonglet's avatar
Cecile Tonglet committed
70
pub use polkadot_cli::Cli;
71
72
pub use polkadot_validation::SignedStatement;
pub use polkadot_primitives::parachain::CollatorId;
73
pub use sc_network::PeerId;
74
pub use service::RuntimeApiCollection;
Cecile Tonglet's avatar
Cecile Tonglet committed
75
pub use sc_cli::SubstrateCli;
76

77
const COLLATION_TIMEOUT: Duration = Duration::from_secs(30);
Gav's avatar
Gav committed
78

79
/// An abstraction over the `Network` with useful functions for a `Collator`.
80
pub trait Network: Send + Sync {
81
82
83
84
85
	/// Create a `Stream` of checked statements for the given `relay_parent`.
	///
	/// The returned stream will not terminate, so it is required to make sure that the stream is
	/// dropped when it is not required anymore. Otherwise, it will stick around in memory
	/// infinitely.
86
	fn checked_statements(&self, relay_parent: Hash) -> Pin<Box<dyn Stream<Item=SignedStatement>>>;
87
88
}

89
90
91
impl Network for polkadot_network::protocol::Service {
	fn checked_statements(&self, relay_parent: Hash) -> Pin<Box<dyn Stream<Item=SignedStatement>>> {
		polkadot_network::protocol::Service::checked_statements(self, relay_parent)
92
93
94
	}
}

95
96
97
98
99
100
/// Error to return when the head data was invalid.
#[derive(Clone, Copy, Debug)]
pub struct InvalidHead;

/// Collation errors.
#[derive(Debug)]
Ashley's avatar
Ashley committed
101
pub enum Error {
102
	/// Error on the relay-chain side of things.
Ashley's avatar
Ashley committed
103
	Polkadot(String),
104
105
106
107
	/// Error on the collator side of things.
	Collator(InvalidHead),
}

Ashley's avatar
Ashley committed
108
impl fmt::Display for Error {
109
110
111
112
113
114
115
116
	fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
		match *self {
			Error::Polkadot(ref err) => write!(f, "Polkadot node error: {}", err),
			Error::Collator(_) => write!(f, "Collator node error: Invalid head data"),
		}
	}
}

117
/// The Polkadot client type.
118
pub type PolkadotClient<B, E, R> = sc_client::Client<B, E, Block, R>;
119

120
121
122
123
124
125
/// Something that can build a `ParachainContext`.
pub trait BuildParachainContext {
	/// The parachain context produced by the `build` function.
	type ParachainContext: self::ParachainContext;

	/// Build the `ParachainContext`.
126
	fn build<B, E, R, SP, Extrinsic>(
127
		self,
128
		client: Arc<PolkadotClient<B, E, R>>,
129
		spawner: SP,
130
		network: impl Network + Clone + 'static,
131
132
	) -> Result<Self::ParachainContext, ()>
		where
133
			PolkadotClient<B, E, R>: ProvideRuntimeApi<Block>,
134
			<PolkadotClient<B, E, R> as ProvideRuntimeApi<Block>>::Api: RuntimeApiCollection<Extrinsic>,
135
136
			// Rust bug: https://github.com/rust-lang/rust/issues/24159
			<<PolkadotClient<B, E, R> as ProvideRuntimeApi<Block>>::Api as sp_api::ApiExt<Block>>::StateBackend:
Gavin Wood's avatar
Gavin Wood committed
137
				sp_api::StateBackend<BlakeTwo256>,
138
139
			Extrinsic: codec::Codec + Send + Sync + 'static,
			E: sc_client::CallExecutor<Block> + Clone + Send + Sync + 'static,
140
141
142
143
			SP: Spawn + Clone + Send + Sync + 'static,
			R: Send + Sync + 'static,
			B: sc_client_api::Backend<Block> + 'static,
			// Rust bug: https://github.com/rust-lang/rust/issues/24159
Gavin Wood's avatar
Gavin Wood committed
144
			B::State: sp_api::StateBackend<BlakeTwo256>;
145
146
}

Gav's avatar
Gav committed
147
148
149
/// Parachain context needed for collation.
///
/// This can be implemented through an externally attached service or a stub.
150
151
/// This is expected to be a lightweight, shared type like an Arc.
pub trait ParachainContext: Clone {
Ashley's avatar
Ashley committed
152
	type ProduceCandidate: Future<Output = Result<(BlockData, HeadData), InvalidHead>>;
153

154
155
	/// Produce a candidate, given the relay parent hash, the latest ingress queue information
	/// and the last parachain head.
Ashley's avatar
Ashley committed
156
	fn produce_candidate(
157
		&mut self,
158
		relay_parent: Hash,
159
		status: LocalValidationData,
160
	) -> Self::ProduceCandidate;
Gav's avatar
Gav committed
161
162
}

163
/// Produce a candidate for the parachain, with given contexts, parent head, and signing key.
Ashley's avatar
Ashley committed
164
pub async fn collate<P>(
165
	relay_parent: Hash,
166
	local_id: ParaId,
167
	local_validation_data: LocalValidationData,
168
	mut para_context: P,
169
	key: Arc<CollatorPair>,
170
)
Ashley's avatar
Ashley committed
171
	-> Result<parachain::Collation, Error>
Gav's avatar
Gav committed
172
	where
173
174
		P: ParachainContext,
		P::ProduceCandidate: Send,
Gav's avatar
Gav committed
175
{
Ashley's avatar
Ashley committed
176
	let (block_data, head_data) = para_context.produce_candidate(
177
		relay_parent,
178
		local_validation_data,
179
180
	).map_err(Error::Collator).await?;

181
182
183
184
185
186
187
188
189
190
191
	let pov_block = PoVBlock {
		block_data,
	};

	let pov_block_hash = pov_block.hash();
	let signature = key.sign(&parachain::collator_signature_payload(
		&relay_parent,
		&local_id,
		&pov_block_hash,
	));

192
	let info = parachain::CollationInfo {
193
		parachain_index: local_id,
194
		relay_parent,
195
196
		collator: key.public(),
		signature,
197
		head_data,
198
		pov_block_hash,
199
200
201
	};

	let collation = parachain::Collation {
202
		info,
203
		pov: pov_block,
204
205
	};

Ashley's avatar
Ashley committed
206
	Ok(collation)
207
208
}

209
fn build_collator_service<S, P, Extrinsic>(
210
	service: (S, polkadot_service::FullNodeHandles),
211
	para_id: ParaId,
212
	key: Arc<CollatorPair>,
213
	build_parachain_context: P,
Gavin Wood's avatar
Gavin Wood committed
214
) -> Result<S, polkadot_service::Error>
215
	where
216
		S: AbstractService<Block = service::Block>,
217
218
		sc_client::Client<S::Backend, S::CallExecutor, service::Block, S::RuntimeApi>: ProvideRuntimeApi<Block>,
		<sc_client::Client<S::Backend, S::CallExecutor, service::Block, S::RuntimeApi> as ProvideRuntimeApi<Block>>::Api:
219
			RuntimeApiCollection<
220
221
222
223
				Extrinsic,
				Error = sp_blockchain::Error,
				StateBackend = sc_client_api::StateBackendFor<S::Backend, Block>
			>,
224
		// Rust bug: https://github.com/rust-lang/rust/issues/24159
225
		S::Backend: service::Backend<service::Block>,
226
		// Rust bug: https://github.com/rust-lang/rust/issues/24159
227
		<S::Backend as service::Backend<service::Block>>::State:
Gavin Wood's avatar
Gavin Wood committed
228
			sp_api::StateBackend<sp_runtime::traits::HashFor<Block>>,
229
230
		// Rust bug: https://github.com/rust-lang/rust/issues/24159
		S::CallExecutor: service::CallExecutor<service::Block>,
231
232
233
234
235
		// Rust bug: https://github.com/rust-lang/rust/issues/24159
		S::SelectChain: service::SelectChain<service::Block>,
		P: BuildParachainContext,
		P::ParachainContext: Send + 'static,
		<P::ParachainContext as ParachainContext>::ProduceCandidate: Send,
236
		Extrinsic: service::Codec + Send + Sync + 'static,
237
{
238
	let (service, handles) = service;
239
	let spawner = service.spawn_task_handle();
240

241
242
243
244
245
	let polkadot_network = match handles.polkadot_network {
		None => return Err(
			"Collator cannot run when Polkadot-specific networking has not been started".into()
		),
		Some(n) => n,
246
	};
247

248
	let client = service.client();
249
250
251
252

	let parachain_context = match build_parachain_context.build(
		client.clone(),
		spawner,
253
		polkadot_network.clone(),
254
255
256
	) {
		Ok(ctx) => ctx,
		Err(()) => {
Gavin Wood's avatar
Gavin Wood committed
257
			return Err("Could not build the parachain context!".into())
258
259
		}
	};
260

261
262
263
264
	let work = async move {
		let mut notification_stream = client.import_notification_stream();

		while let Some(notification) = notification_stream.next().await {
265
266
267
268
			macro_rules! try_fr {
				($e:expr) => {
					match $e {
						Ok(x) => x,
269
						Err(e) => return future::Either::Left(future::err(Error::Polkadot(
270
271
							format!("{:?}", e)
						))),
272
					}
273
				}
274
			}
275

276
277
278
			let relay_parent = notification.hash;
			let id = BlockId::hash(relay_parent);

279
			let network = polkadot_network.clone();
280
281
282
283
			let client = client.clone();
			let key = key.clone();
			let parachain_context = parachain_context.clone();

284
			let work = future::lazy(move |_| {
285
				let api = client.runtime_api();
286
287
				let local_validation = match try_fr!(api.local_validation_data(&id, para_id)) {
					Some(local_validation) => local_validation,
288
					None => return future::Either::Left(future::ok(())),
289
290
291
292
293
294
295
296
297
298
				};

				let validators = try_fr!(api.validators(&id));

				let targets = compute_targets(
					para_id,
					validators.as_slice(),
					try_fr!(api.duty_roster(&id)),
				);

299
				let collation_work = collate(
300
301
					relay_parent,
					para_id,
302
					local_validation,
303
304
					parachain_context,
					key,
Ashley's avatar
Ashley committed
305
				).map_ok(move |collation| {
306
					network.distribute_collation(targets, collation)
307
308
309
				});

				future::Either::Right(collation_work)
310
			});
311

312
			let deadlined = future::select(
313
				work.then(|f| f).boxed(),
314
315
316
317
318
319
320
321
322
				futures_timer::Delay::new(COLLATION_TIMEOUT)
			);

			let silenced = deadlined
				.map(|either| {
					if let future::Either::Right(_) = either {
						warn!("Collation failure: timeout");
					}
				});
323

324
			let future = silenced.map(drop);
325

326
			tokio::spawn(future);
327
328
		}
	}.boxed();
329

Gavin Wood's avatar
Gavin Wood committed
330
	service.spawn_essential_task("collation", work);
331

Gavin Wood's avatar
Gavin Wood committed
332
	Ok(service)
333
334
}

335
336
337
338
339
340
341
342
343
344
345
346
347
/// Async function that will run the collator node with the given `RelayChainContext` and `ParachainContext`
/// built by the given `BuildParachainContext` and arguments to the underlying polkadot node.
pub async fn start_collator<P>(
	build_parachain_context: P,
	para_id: ParaId,
	key: Arc<CollatorPair>,
	config: Configuration,
) -> Result<(), polkadot_service::Error>
where
	P: BuildParachainContext,
	P::ParachainContext: Send + 'static,
	<P::ParachainContext as ParachainContext>::ProduceCandidate: Send,
{
Cecile Tonglet's avatar
WIP    
Cecile Tonglet committed
348
	let is_kusama = config.chain_spec.is_kusama();
349
350
	match (is_kusama, &config.role) {
		(_, Role::Light) => return Err(
351
352
			polkadot_service::Error::Other("light nodes are unsupported as collator".into())
		).into(),
353
354
		(true, _) =>
			build_collator_service(
355
				service::kusama_new_full(config, Some((key.public(), para_id)), None, false, 6000, None)?,
356
357
358
359
360
361
				para_id,
				key,
				build_parachain_context,
			)?.await,
		(false, _) =>
			build_collator_service(
362
				service::polkadot_new_full(config, Some((key.public(), para_id)), None, false, 6000, None)?,
363
364
365
366
367
368
369
				para_id,
				key,
				build_parachain_context,
			)?.await,
	}
}

370
fn compute_targets(para_id: ParaId, session_keys: &[ValidatorId], roster: DutyRoster) -> HashSet<ValidatorId> {
371
372
373
374
375
376
377
378
379
	use polkadot_primitives::parachain::Chain;

	roster.validator_duty.iter().enumerate()
		.filter(|&(_, c)| c == &Chain::Parachain(para_id))
		.filter_map(|(i, _)| session_keys.get(i))
		.cloned()
		.collect()
}

Gav's avatar
Gav committed
380
381
382
383
#[cfg(test)]
mod tests {
	use super::*;

384
385
386
387
	#[derive(Clone)]
	struct DummyParachainContext;

	impl ParachainContext for DummyParachainContext {
Ashley's avatar
Ashley committed
388
		type ProduceCandidate = future::Ready<Result<(BlockData, HeadData), InvalidHead>>;
389

Ashley's avatar
Ashley committed
390
		fn produce_candidate(
391
			&mut self,
392
			_relay_parent: Hash,
393
			_local_validation: LocalValidationData,
394
		) -> Self::ProduceCandidate {
395
			// send messages right back.
396
			future::ok((
397
398
399
				BlockData(vec![1, 2, 3, 4, 5,]),
				HeadData(vec![9, 9, 9]),
			))
Gav's avatar
Gav committed
400
401
		}
	}
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422

	struct BuildDummyParachainContext;

	impl BuildParachainContext for BuildDummyParachainContext {
		type ParachainContext = DummyParachainContext;

		fn build<B, E, R, SP, Extrinsic>(
			self,
			_: Arc<PolkadotClient<B, E, R>>,
			_: SP,
			_: impl Network + Clone + 'static,
		) -> Result<Self::ParachainContext, ()> {
			Ok(DummyParachainContext)
		}
	}

	// Make sure that the future returned by `start_collator` implementes `Send`.
	#[test]
	fn start_collator_is_send() {
		fn check_send<T: Send>(_: T) {}

Cecile Tonglet's avatar
Cecile Tonglet committed
423
424
425
426
		let cli = Cli::from_iter(&["-dev"]);
		let task_executor = Arc::new(|_| unimplemented!());
		let config = cli.create_configuration(&cli.run.base, task_executor).unwrap();

427
428
429
430
		check_send(start_collator(
			BuildDummyParachainContext,
			0.into(),
			Arc::new(CollatorPair::generate().0),
Cecile Tonglet's avatar
Cecile Tonglet committed
431
			config,
432
433
		));
	}
Gav's avatar
Gav committed
434
}