where
P: PolkadotApi + Send + Sync + 'static
{
/// The client instance.
pub client: Arc,
/// The transaction pool.
pub transaction_pool: Arc>,
/// The backing network handle.
pub network: N,
/// Parachain collators.
pub collators: C,
/// handle to remote task executor
pub handle: TaskExecutor,
/// The duration after which parachain-empty blocks will be allowed.
pub parachain_empty_duration: Duration,
/// Store for extrinsic data.
pub extrinsic_store: ExtrinsicStore,
/// Offline-tracker.
pub offline: SharedOfflineTracker,
}
impl bft::Environment for ProposerFactory
where
C: Collators + Send + 'static,
N: Network,
P: PolkadotApi + Send + Sync + 'static,
::Future: Send + 'static,
N::TableRouter: Send + 'static,
{
type Proposer = Proposer;
type Input = N::Input;
type Output = N::Output;
type Error = Error;
fn init(
&self,
parent_header: &Header,
authorities: &[AuthorityId],
sign_with: Arc,
) -> Result<(Self::Proposer, Self::Input, Self::Output), Error> {
use runtime_primitives::traits::{Hash as HashT, BlakeTwo256};
// force delay in evaluation this long.
const FORCE_DELAY: Timestamp = 5;
let parent_hash = parent_header.hash().into();
let id = BlockId::hash(parent_hash);
let duty_roster = self.client.duty_roster(&id)?;
let random_seed = self.client.random_seed(&id)?;
let random_seed = BlakeTwo256::hash(&*random_seed);
let validators = self.client.validators(&id)?;
self.offline.write().note_new_block(&validators[..]);
let (group_info, local_duty) = make_group_info(
duty_roster,
authorities,
sign_with.public().into(),
)?;
info!("Starting consensus session on top of parent {:?}. Local parachain duty is {:?}",
parent_hash, local_duty.validation);
let active_parachains = self.client.active_parachains(&id)?;
debug!(target: "consensus", "Active parachains: {:?}", active_parachains);
let n_parachains = active_parachains.len();
let table = Arc::new(SharedTable::new(group_info, sign_with.clone(), parent_hash, self.extrinsic_store.clone()));
let (router, input, output) = self.network.communication_for(
authorities,
table.clone(),
self.handle.clone()
);
let now = Instant::now();
let dynamic_inclusion = DynamicInclusion::new(
n_parachains,
now,
self.parachain_empty_duration.clone(),
);
let validation_para = match local_duty.validation {
Chain::Relay => None,
Chain::Parachain(id) => Some(id),
};
let collation_work = validation_para.map(|para| CollationFetch::new(
para,
id.clone(),
parent_hash.clone(),
self.collators.clone(),
self.client.clone(),
));
let drop_signal = dispatch_collation_work(
router.clone(),
&self.handle,
collation_work,
self.extrinsic_store.clone(),
);
let proposer = Proposer {
client: self.client.clone(),
dynamic_inclusion,
local_key: sign_with,
parent_hash,
parent_id: id,
parent_number: parent_header.number,
random_seed,
table,
transaction_pool: self.transaction_pool.clone(),
offline: self.offline.clone(),
validators,
minimum_timestamp: current_timestamp() + FORCE_DELAY,
_drop_signal: drop_signal,
};
Ok((proposer, input, output))
}
}
// dispatch collation work to be done in the background. returns a signal object
// that should fire when the collation work is no longer necessary (e.g. when the proposer object is dropped)
fn dispatch_collation_work(
router: R,
handle: &TaskExecutor,
work: Option>,
extrinsic_store: ExtrinsicStore,
) -> exit_future::Signal where
C: Collators + Send + 'static,
P: PolkadotApi + Send + Sync + 'static,
::Future: Send + 'static,
R: TableRouter + Send + 'static,
{
use extrinsic_store::Data;
let (signal, exit) = exit_future::signal();
let work = match work {
Some(w) => w,
None => return signal,
};
let relay_parent = work.relay_parent();
let handled_work = work.then(move |result| match result {
Ok((collation, extrinsic)) => {
let res = extrinsic_store.make_available(Data {
relay_parent,
parachain_id: collation.receipt.parachain_index,
candidate_hash: collation.receipt.hash(),
block_data: collation.block_data.clone(),
extrinsic: Some(extrinsic.clone()),
});
match res {
Ok(()) =>
router.local_candidate(collation.receipt, collation.block_data, extrinsic),
Err(e) =>
warn!(target: "consensus", "Failed to make collation data available: {:?}", e),
}
Ok(())
}
Err(_e) => {
warn!(target: "consensus", "Failed to collate candidate");
Ok(())
}
});
let cancellable_work = handled_work.select(exit).then(|_| Ok(()));
// spawn onto thread pool.
handle.spawn(cancellable_work);
signal
}
struct LocalDuty {
validation: Chain,
}
/// The Polkadot proposer logic.
pub struct Proposer {
client: Arc,
dynamic_inclusion: DynamicInclusion,
local_key: Arc,
parent_hash: Hash,
parent_id: BlockId,
parent_number: BlockNumber,
random_seed: Hash,
table: Arc,
transaction_pool: Arc>,
offline: SharedOfflineTracker,
validators: Vec,
minimum_timestamp: u64,
_drop_signal: exit_future::Signal,
}
impl Proposer {
fn primary_index(&self, round_number: usize, len: usize) -> usize {
use primitives::uint::U256;
let big_len = U256::from(len);
let offset = U256::from_big_endian(&self.random_seed.0) % big_len;
let offset = offset.low_u64() as usize + round_number;
offset % len
}
}
impl bft::Proposer for Proposer
where
C: PolkadotApi + Send + Sync,
{
type Error = Error;
type Create = future::Either<
CreateProposal,
future::FutureResult,
>;
type Evaluate = Box>;
fn propose(&self) -> Self::Create {
const ATTEMPT_PROPOSE_EVERY: Duration = Duration::from_millis(100);
let initial_included = self.table.includable_count();
let now = Instant::now();
let enough_candidates = self.dynamic_inclusion.acceptable_in(
now,
initial_included,
).unwrap_or_else(|| now + Duration::from_millis(1));
let timing = ProposalTiming {
attempt_propose: Interval::new(now + ATTEMPT_PROPOSE_EVERY, ATTEMPT_PROPOSE_EVERY),
enough_candidates: Delay::new(enough_candidates),
dynamic_inclusion: self.dynamic_inclusion.clone(),
last_included: initial_included,
};
future::Either::A(CreateProposal {
parent_hash: self.parent_hash.clone(),
parent_number: self.parent_number.clone(),
parent_id: self.parent_id.clone(),
client: self.client.clone(),
transaction_pool: self.transaction_pool.clone(),
table: self.table.clone(),
offline: self.offline.clone(),
validators: self.validators.clone(),
minimum_timestamp: self.minimum_timestamp,
timing,
})
}
fn evaluate(&self, unchecked_proposal: &Block) -> Self::Evaluate {
debug!(target: "bft", "evaluating block on top of parent ({}, {:?})", self.parent_number, self.parent_hash);
let active_parachains = match self.client.active_parachains(&self.parent_id) {
Ok(x) => x,
Err(e) => return Box::new(future::err(e.into())) as Box<_>,
};
let current_timestamp = current_timestamp();
// do initial serialization and structural integrity checks.
let maybe_proposal = evaluation::evaluate_initial(
unchecked_proposal,
current_timestamp,
&self.parent_hash,
self.parent_number,
&active_parachains,
);
let proposal = match maybe_proposal {
Ok(p) => p,
Err(e) => {
// TODO: these errors are easily re-checked in runtime.
debug!(target: "bft", "Invalid proposal: {:?}", e);
return Box::new(future::ok(false));
}
};
let vote_delays = {
let now = Instant::now();
let included_candidate_hashes = proposal
.parachain_heads()
.iter()
.map(|candidate| candidate.hash());
// delay casting vote until we have proof that all candidates are
// includable.
let includability_tracker = self.table.track_includability(included_candidate_hashes)
.map_err(|_| ErrorKind::PrematureDestruction.into());
// the duration at which the given number of parachains is acceptable.
let count_delay = self.dynamic_inclusion.acceptable_in(
now,
proposal.parachain_heads().len(),
);
// the duration until the given timestamp is current
let proposed_timestamp = ::std::cmp::max(self.minimum_timestamp, proposal.timestamp());
let timestamp_delay = if proposed_timestamp > current_timestamp {
let delay_s = proposed_timestamp - current_timestamp;
debug!(target: "bft", "Delaying evaluation of proposal for {} seconds", delay_s);
Some(now + Duration::from_secs(delay_s))
} else {
None
};
// delay casting vote until able according to minimum block time,
// timestamp delay, and count delay.
// construct a future from the maximum of the two durations.
let max_delay = ::std::cmp::max(timestamp_delay, count_delay);
let temporary_delay = match max_delay {
Some(duration) => future::Either::A(
Delay::new(duration).map_err(|e| Error::from(ErrorKind::Timer(e)))
),
None => future::Either::B(future::ok(())),
};
includability_tracker.join(temporary_delay)
};
// refuse to vote if this block says a validator is offline that we
// think isn't.
let offline = proposal.noted_offline();
if !self.offline.read().check_consistency(&self.validators[..], offline) {
return Box::new(futures::empty());
}
// evaluate whether the block is actually valid.
// TODO: is it better to delay this until the delays are finished?
let evaluated = self.client
.evaluate_block(&self.parent_id, unchecked_proposal.clone())
.map_err(Into::into);
let future = future::result(evaluated).and_then(move |good| {
let end_result = future::ok(good);
if good {
// delay a "good" vote.
future::Either::A(vote_delays.and_then(|_| end_result))
} else {
// don't delay a "bad" evaluation.
future::Either::B(end_result)
}
});
Box::new(future) as Box<_>
}
fn round_proposer(&self, round_number: usize, authorities: &[AuthorityId]) -> AuthorityId {
let offset = self.primary_index(round_number, authorities.len());
let proposer = authorities[offset].clone();
trace!(target: "bft", "proposer for round {} is {}", round_number, proposer);
proposer
}
fn import_misbehavior(&self, misbehavior: Vec<(AuthorityId, bft::Misbehavior)>) {
use rhododendron::Misbehavior as GenericMisbehavior;
use runtime_primitives::bft::{MisbehaviorKind, MisbehaviorReport};
use runtime_primitives::MaybeUnsigned;
use polkadot_runtime::{Call, Extrinsic, BareExtrinsic, UncheckedExtrinsic, ConsensusCall};
let local_id = self.local_key.public().0.into();
let mut next_index = {
let cur_index = self.transaction_pool.cull_and_get_pending(&BlockId::hash(self.parent_hash), |pending| pending
.filter(|tx| tx.verified.sender().map(|s| s == local_id).unwrap_or(false))
.last()
.map(|tx| Ok(tx.verified.index()))
.unwrap_or_else(|| self.client.index(&self.parent_id, local_id))
);
match cur_index {
Ok(Ok(cur_index)) => cur_index + 1,
Ok(Err(e)) => {
warn!(target: "consensus", "Error computing next transaction index: {}", e);
return;
}
Err(e) => {
warn!(target: "consensus", "Error computing next transaction index: {}", e);
return;
}
}
};
for (target, misbehavior) in misbehavior {
let report = MisbehaviorReport {
parent_hash: self.parent_hash,
parent_number: self.parent_number,
target,
misbehavior: match misbehavior {
GenericMisbehavior::ProposeOutOfTurn(_, _, _) => continue,
GenericMisbehavior::DoublePropose(_, _, _) => continue,
GenericMisbehavior::DoublePrepare(round, (h1, s1), (h2, s2))
=> MisbehaviorKind::BftDoublePrepare(round as u32, (h1, s1.signature), (h2, s2.signature)),
GenericMisbehavior::DoubleCommit(round, (h1, s1), (h2, s2))
=> MisbehaviorKind::BftDoubleCommit(round as u32, (h1, s1.signature), (h2, s2.signature)),
}
};
let extrinsic = BareExtrinsic {
signed: local_id,
index: next_index,
function: Call::Consensus(ConsensusCall::report_misbehavior(report)),
};
next_index += 1;
let signature = MaybeUnsigned(self.local_key.sign(&extrinsic.encode()).into());
let extrinsic = Extrinsic {
signed: extrinsic.signed.into(),
index: extrinsic.index,
function: extrinsic.function,
};
let uxt: Vec = Decode::decode(&mut UncheckedExtrinsic::new(extrinsic, signature).encode().as_slice()).expect("Encoded extrinsic is valid");
self.transaction_pool.submit_one(&BlockId::hash(self.parent_hash), uxt)
.expect("locally signed extrinsic is valid; qed");
}
}
fn on_round_end(&self, round_number: usize, was_proposed: bool) {
let primary_validator = self.validators[
self.primary_index(round_number, self.validators.len())
];
// alter the message based on whether we think the empty proposer was forced to skip the round.
// this is determined by checking if our local validator would have been forced to skip the round.
let consider_online = was_proposed || {
let forced_delay = self.dynamic_inclusion.acceptable_in(Instant::now(), self.table.includable_count());
let public = ::ed25519::Public::from_raw(primary_validator.0);
match forced_delay {
None => info!(
"Potential Offline Validator: {} failed to propose during assigned slot: {}",
public,
round_number,
),
Some(_) => info!(
"Potential Offline Validator {} potentially forced to skip assigned slot: {}",
public,
round_number,
),
}
forced_delay.is_some()
};
self.offline.write().note_round_end(primary_validator, consider_online);
}
}
fn current_timestamp() -> Timestamp {
time::SystemTime::now().duration_since(time::UNIX_EPOCH)
.expect("now always later than unix epoch; qed")
.as_secs()
}
struct ProposalTiming {
attempt_propose: Interval,
dynamic_inclusion: DynamicInclusion,
enough_candidates: Delay,
last_included: usize,
}
impl ProposalTiming {
// whether it's time to attempt a proposal.
// shouldn't be called outside of the context of a task.
fn poll(&mut self, included: usize) -> Poll<(), ErrorKind> {
// first drain from the interval so when the minimum delay is up
// we don't have any notifications built up.
//
// this interval is just meant to produce periodic task wakeups
// that lead to the `dynamic_inclusion` getting updated as necessary.
if let Async::Ready(x) = self.attempt_propose.poll().map_err(ErrorKind::Timer)? {
x.expect("timer still alive; intervals never end; qed");
}
if included == self.last_included {
return self.enough_candidates.poll().map_err(ErrorKind::Timer);
}
// the amount of includable candidates has changed. schedule a wakeup
// if it's not sufficient anymore.
match self.dynamic_inclusion.acceptable_in(Instant::now(), included) {
Some(instant) => {
self.last_included = included;
self.enough_candidates.reset(instant);
self.enough_candidates.poll().map_err(ErrorKind::Timer)
}
None => Ok(Async::Ready(())),
}
}
}
/// Future which resolves upon the creation of a proposal.
pub struct CreateProposal {
parent_hash: Hash,
parent_number: BlockNumber,
parent_id: BlockId,
client: Arc,
transaction_pool: Arc>,
table: Arc,
timing: ProposalTiming,
validators: Vec,
offline: SharedOfflineTracker,
minimum_timestamp: Timestamp,
}
impl CreateProposal where C: PolkadotApi + Send + Sync {
fn propose_with(&self, candidates: Vec) -> Result {
use polkadot_api::BlockBuilder;
use runtime_primitives::traits::{Hash as HashT, BlakeTwo256};
use polkadot_primitives::InherentData;
const MAX_VOTE_OFFLINE_SECONDS: Duration = Duration::from_secs(60);
// TODO: handle case when current timestamp behind that in state.
let timestamp = ::std::cmp::max(self.minimum_timestamp, current_timestamp());
let elapsed_since_start = self.timing.dynamic_inclusion.started_at().elapsed();
let offline_indices = if elapsed_since_start > MAX_VOTE_OFFLINE_SECONDS {
Vec::new()
} else {
self.offline.read().reports(&self.validators[..])
};
if !offline_indices.is_empty() {
info!(
"Submitting offline validators {:?} for slash-vote",
offline_indices.iter().map(|&i| self.validators[i as usize]).collect::>(),
)
}
let inherent_data = InherentData {
timestamp,
parachain_heads: candidates,
offline_indices,
};
let mut block_builder = self.client.build_block(&self.parent_id, inherent_data)?;
{
let mut unqueue_invalid = Vec::new();
let result = self.transaction_pool.cull_and_get_pending(&BlockId::hash(self.parent_hash), |pending_iterator| {
let mut pending_size = 0;
for pending in pending_iterator {
if pending_size + pending.verified.encoded_size() >= MAX_TRANSACTIONS_SIZE { break }
match block_builder.push_extrinsic(pending.original.clone()) {
Ok(()) => {
pending_size += pending.verified.encoded_size();
}
Err(e) => {
trace!(target: "transaction-pool", "Invalid transaction: {}", e);
unqueue_invalid.push(pending.verified.hash().clone());
}
}
}
});
if let Err(e) = result {
warn!("Unable to get the pending set: {:?}", e);
}
self.transaction_pool.remove(&unqueue_invalid, false);
}
let polkadot_block = block_builder.bake()?;
info!("Proposing block [number: {}; hash: {}; parent_hash: {}; extrinsics: [{}]]",
polkadot_block.header.number,
Hash::from(polkadot_block.header.hash()),
polkadot_block.header.parent_hash,
polkadot_block.extrinsics.iter()
.map(|xt| format!("{}", BlakeTwo256::hash_of(xt)))
.collect::>()
.join(", ")
);
let substrate_block = Decode::decode(&mut polkadot_block.encode().as_slice())
.expect("polkadot blocks defined to serialize to substrate blocks correctly; qed");
// TODO: full re-evaluation
let active_parachains = self.client.active_parachains(&self.parent_id)?;
assert!(evaluation::evaluate_initial(
&substrate_block,
timestamp,
&self.parent_hash,
self.parent_number,
&active_parachains,
).is_ok());
Ok(substrate_block)
}
}
impl Future for CreateProposal where C: PolkadotApi + Send + Sync {
type Item = Block;
type Error = Error;
fn poll(&mut self) -> Poll {
// 1. try to propose if we have enough includable candidates and other
// delays have concluded.
let included = self.table.includable_count();
try_ready!(self.timing.poll(included));
// 2. propose
let proposed_candidates = self.table.with_proposal(|proposed_set| {
proposed_set.into_iter().cloned().collect()
});
self.propose_with(proposed_candidates).map(Async::Ready)
}
}
#[cfg(test)]
mod tests {
use super::*;
use substrate_keyring::Keyring;
#[test]
fn sign_and_check_statement() {
let statement: Statement = GenericStatement::Valid([1; 32].into());
let parent_hash = [2; 32].into();
let sig = sign_table_statement(&statement, &Keyring::Alice.pair(), &parent_hash);
assert!(check_statement(&statement, &sig, Keyring::Alice.to_raw_public().into(), &parent_hash));
assert!(!check_statement(&statement, &sig, Keyring::Alice.to_raw_public().into(), &[0xff; 32].into()));
assert!(!check_statement(&statement, &sig, Keyring::Bob.to_raw_public().into(), &parent_hash));
}
}