#![deny(missing_docs)]
#![deny(unused_crate_dependencies)]
#![recursion_limit = "256"]
use std::time::{Duration, Instant};
use futures::{
stream::{FusedStream, StreamExt},
FutureExt, TryFutureExt,
};
use polkadot_node_subsystem_util::reputation::ReputationAggregator;
use sp_keystore::KeystorePtr;
use polkadot_node_network_protocol::{
request_response::{v1 as request_v1, v2 as protocol_v2, IncomingRequestReceiver},
PeerId, UnifiedReputationChange as Rep,
};
use polkadot_primitives::CollatorPair;
use polkadot_node_subsystem::{errors::SubsystemError, overseer, DummySubsystem, SpawnedSubsystem};
mod error;
mod collator_side;
mod validator_side;
const LOG_TARGET: &'static str = "parachain::collator-protocol";
#[derive(Debug, Clone, Copy)]
pub struct CollatorEvictionPolicy {
pub inactive_collator: Duration,
pub undeclared: Duration,
}
impl Default for CollatorEvictionPolicy {
fn default() -> Self {
CollatorEvictionPolicy {
inactive_collator: Duration::from_secs(24),
undeclared: Duration::from_secs(1),
}
}
}
pub enum ProtocolSide {
Validator {
keystore: KeystorePtr,
eviction_policy: CollatorEvictionPolicy,
metrics: validator_side::Metrics,
},
Collator {
peer_id: PeerId,
collator_pair: CollatorPair,
request_receiver_v1: IncomingRequestReceiver<request_v1::CollationFetchingRequest>,
request_receiver_v2: IncomingRequestReceiver<protocol_v2::CollationFetchingRequest>,
metrics: collator_side::Metrics,
},
None,
}
pub struct CollatorProtocolSubsystem {
protocol_side: ProtocolSide,
}
#[overseer::contextbounds(CollatorProtocol, prefix = self::overseer)]
impl CollatorProtocolSubsystem {
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 } =>
validator_side::run(ctx, keystore, eviction_policy, metrics)
.map_err(|e| SubsystemError::with_origin("collator-protocol", e))
.boxed(),
ProtocolSide::Collator {
peer_id,
collator_pair,
request_receiver_v1,
request_receiver_v2,
metrics,
} => collator_side::run(
ctx,
peer_id,
collator_pair,
request_receiver_v1,
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 }
}
}
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;
}
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()
}
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()
}