polkadot_collator_protocol/

lib.rs

// 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/>.

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

#![deny(missing_docs)]
#![deny(unused_crate_dependencies)]
#![recursion_limit = "256"]

use std::{
	collections::{HashMap, HashSet},
	sync::Arc,
	time::{Duration, Instant},
};

use futures::{
	channel::oneshot,
	stream::{FusedStream, StreamExt},
	FutureExt, TryFutureExt,
};

use polkadot_node_subsystem::CollatorProtocolSenderTrait;
use polkadot_node_subsystem_util::{database::Database, reputation::ReputationAggregator};
use sp_consensus_babe::digests::CompatibleDigestItem;
use sp_core::H256;
use sp_keystore::KeystorePtr;

use polkadot_node_network_protocol::{
	request_response::{v2 as protocol_v2, IncomingRequestReceiver},
	PeerId, UnifiedReputationChange as Rep,
};
use polkadot_node_subsystem::{
	errors::SubsystemError, messages::ChainApiMessage, overseer, DummySubsystem, SpawnedSubsystem,
};
use polkadot_primitives::{CollatorPair, Hash, RELAY_CHAIN_SLOT_DURATION_MILLIS};
use sp_consensus_slots::SlotDuration;
pub use validator_side_experimental::ReputationConfig;

mod collator_side;
mod validator_side;
mod validator_side_experimental;

// TODO: move into validator_side_experimental once `validator_side` is retired
mod validator_side_metrics;

const LOG_TARGET: &'static str = "parachain::collator-protocol";
const LOG_TARGET_STATS: &'static str = "parachain::collator-protocol::stats";

/// A collator eviction policy - how fast to evict collators which are inactive.
#[derive(Debug, Clone, Copy)]
pub struct CollatorEvictionPolicy {
	/// How fast to evict collators who are inactive.
	pub inactive_collator: Duration,
	/// How fast to evict peers which don't declare their para.
	pub undeclared: Duration,
}

impl Default for CollatorEvictionPolicy {
	fn default() -> Self {
		CollatorEvictionPolicy {
			inactive_collator: Duration::from_secs(24),
			undeclared: Duration::from_secs(1),
		}
	}
}

/// What side of the collator protocol is being engaged
pub enum ProtocolSide {
	/// Validators operate on the relay chain.
	Validator {
		/// The keystore holding validator keys.
		keystore: KeystorePtr,
		/// An eviction policy for inactive peers or validators.
		eviction_policy: CollatorEvictionPolicy,
		/// Prometheus metrics for validators.
		metrics: validator_side::Metrics,
		/// List of invulnerable collators which is handled with a priority.
		invulnerables: HashSet<PeerId>,
		/// Override for `HOLD_OFF_DURATION` constant .
		collator_protocol_hold_off: Option<Duration>,
	},
	/// Experimental variant of the validator side. Do not use in production.
	ValidatorExperimental {
		/// The keystore holding validator keys.
		keystore: KeystorePtr,
		/// Prometheus metrics for validators.
		metrics: validator_side_experimental::Metrics,
		/// Database used for reputation house keeping.
		db: Arc<dyn Database>,
		/// Reputation configuration (column number).
		reputation_config: validator_side_experimental::ReputationConfig,
	},
	/// Collators operate on a parachain.
	Collator {
		/// Local peer id.
		peer_id: PeerId,
		/// Parachain collator pair.
		collator_pair: CollatorPair,
		/// Receiver for v2 collation fetching requests.
		request_receiver_v2: IncomingRequestReceiver<protocol_v2::CollationFetchingRequest>,
		/// Metrics.
		metrics: collator_side::Metrics,
	},
	/// No protocol side, just disable it.
	None,
}

/// The collator protocol subsystem.
pub struct CollatorProtocolSubsystem {
	protocol_side: ProtocolSide,
}

#[overseer::contextbounds(CollatorProtocol, prefix = self::overseer)]
impl CollatorProtocolSubsystem {
	/// Start the collator protocol.
	pub fn new(protocol_side: ProtocolSide) -> Self {
		Self { protocol_side }
	}
}

#[overseer::subsystem(CollatorProtocol, error=SubsystemError, prefix=self::overseer)]
impl<Context> CollatorProtocolSubsystem {
	fn start(self, ctx: Context) -> SpawnedSubsystem {
		let future = match self.protocol_side {
			ProtocolSide::Validator {
				keystore,
				eviction_policy,
				metrics,
				invulnerables,
				collator_protocol_hold_off,
			} => {
				gum::trace!(
					target: LOG_TARGET,
					?invulnerables,
					?collator_protocol_hold_off,
					"AH collator protocol params",
				);
				validator_side::run(
					ctx,
					keystore,
					eviction_policy,
					metrics,
					invulnerables,
					collator_protocol_hold_off,
				)
				.map_err(|e| SubsystemError::with_origin("collator-protocol", e))
				.boxed()
			},
			ProtocolSide::ValidatorExperimental { keystore, metrics, db, reputation_config } => {
				validator_side_experimental::run(ctx, keystore, metrics, db, reputation_config)
					.map_err(|e| SubsystemError::with_origin("collator-protocol", e))
					.boxed()
			},
			ProtocolSide::Collator { peer_id, collator_pair, request_receiver_v2, metrics } => {
				collator_side::run(ctx, peer_id, collator_pair, request_receiver_v2, metrics)
					.map_err(|e| SubsystemError::with_origin("collator-protocol", e))
					.boxed()
			},
			ProtocolSide::None => return DummySubsystem.start(ctx),
		};

		SpawnedSubsystem { name: "collator-protocol-subsystem", future }
	}
}

/// Modify the reputation of a peer based on its behavior.
async fn modify_reputation(
	reputation: &mut ReputationAggregator,
	sender: &mut impl overseer::CollatorProtocolSenderTrait,
	peer: PeerId,
	rep: Rep,
) {
	gum::trace!(
		target: LOG_TARGET,
		rep = ?rep,
		peer_id = %peer,
		"reputation change for peer",
	);

	reputation.modify(sender, peer, rep).await;
}

/// Wait until tick and return the timestamp for the following one.
async fn wait_until_next_tick(last_poll: Instant, period: Duration) -> Instant {
	let now = Instant::now();
	let next_poll = last_poll + period;

	if next_poll > now {
		futures_timer::Delay::new(next_poll - now).await
	}

	Instant::now()
}

/// Returns an infinite stream that yields with an interval of `period`.
fn tick_stream(period: Duration) -> impl FusedStream<Item = ()> {
	futures::stream::unfold(Instant::now(), move |next_check| async move {
		Some(((), wait_until_next_tick(next_check, period).await))
	})
	.fuse()
}

/// Scheduling info tracked per active leaf, used for V3 scheduling parent validation.
/// Stores the leaf's BABE slot and parent hash so the validator can determine whether
/// the scheduling parent corresponds to the last finished relay chain slot.
struct LeafSchedulingInfo {
	/// The parent hash of the leaf block.
	parent_hash: Hash,
	/// The BABE slot of the leaf block.
	slot: sp_consensus_slots::Slot,
}

pub(crate) async fn extract_leaf_scheduling_info<Sender: CollatorProtocolSenderTrait>(
	sender: &mut Sender,
	leaf: H256,
) -> Option<LeafSchedulingInfo> {
	// Fetch leaf header to extract BABE slot for V3 scheduling parent validation.
	// Without this info, V3 advertisements referencing this leaf will be rejected.
	let (tx, rx) = oneshot::channel();
	sender.send_message(ChainApiMessage::BlockHeader(leaf, tx)).await;
	let header = rx.await.ok().and_then(|r| r.ok().flatten());
	header.and_then(|header| {
		let slot = header.digest.logs().iter().find_map(|log| log.as_babe_pre_digest())?.slot();
		Some(LeafSchedulingInfo { parent_hash: header.parent_hash, slot })
	})
}

pub(crate) fn is_scheduling_parent_valid(
	scheduling_parent: &Hash,
	leaf_scheduling_info: &HashMap<Hash, LeafSchedulingInfo>,
) -> bool {
	let slot_duration = SlotDuration::from_millis(RELAY_CHAIN_SLOT_DURATION_MILLIS);
	let current_slot =
		sp_consensus_slots::Slot::from_timestamp(sp_timestamp::Timestamp::current(), slot_duration);
	if let Some(info) = leaf_scheduling_info.get(scheduling_parent) {
		// scheduling_parent is a leaf. This is allowed only when the leaf's slot is
		// the previous slot.
		*current_slot == *info.slot + 1
	} else {
		// scheduling_parent is not a leaf. This is allowed only if the sp is the parent of
		// any leaf whose slot is still in progress.
		leaf_scheduling_info
			.iter()
			.any(|(_, info)| *current_slot == *info.slot && *scheduling_parent == info.parent_hash)
	}
}