// 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 .
//! Implements a `CandidateBackingSubsystem`.
use std::collections::{HashMap, HashSet};
use std::convert::TryFrom;
use std::pin::Pin;
use std::sync::Arc;
use bitvec::vec::BitVec;
use futures::{
channel::{mpsc, oneshot},
task::SpawnError,
Future, FutureExt, SinkExt, StreamExt,
};
use keystore::KeyStorePtr;
use polkadot_primitives::v1::{
CommittedCandidateReceipt, BackedCandidate, Id as ParaId, ValidatorId,
ValidatorIndex, SigningContext, PoV, OmittedValidationData,
CandidateDescriptor, AvailableData, ErasureChunk, ValidatorSignature, Hash, CandidateReceipt,
CandidateCommitments,
};
use polkadot_node_primitives::{
FromTableMisbehavior, Statement, SignedFullStatement, MisbehaviorReport,
ValidationOutputs, ValidationResult,
};
use polkadot_subsystem::{
messages::{
AllMessages, AvailabilityStoreMessage, CandidateBackingMessage, CandidateSelectionMessage,
CandidateValidationMessage, NewBackedCandidate, PoVDistributionMessage, ProvisionableData,
ProvisionerMessage, RuntimeApiMessage, RuntimeApiRequest, StatementDistributionMessage, ValidationFailed,
},
util::{
self,
JobTrait,
JobTraitExt,
Validator,
},
};
use statement_table::{
generic::AttestedCandidate as TableAttestedCandidate,
Context as TableContextTrait,
Table,
v1::{
Statement as TableStatement,
SignedStatement as TableSignedStatement, Summary as TableSummary,
},
};
#[derive(Debug, derive_more::From)]
pub enum Error {
CandidateNotFound,
InvalidSignature,
#[from]
Erasure(erasure_coding::Error),
#[from]
ValidationFailed(ValidationFailed),
#[from]
Oneshot(oneshot::Canceled),
#[from]
Mpsc(mpsc::SendError),
#[from]
Spawn(SpawnError),
#[from]
UtilError(util::Error),
}
/// Holds all data needed for candidate backing job operation.
pub struct CandidateBackingJob {
/// The hash of the relay parent on top of which this job is doing it's work.
parent: Hash,
/// Inbound message channel receiving part.
rx_to: mpsc::Receiver,
/// Outbound message channel sending part.
tx_from: mpsc::Sender,
/// The `ParaId`s assigned to this validator.
assignment: ParaId,
/// We issued `Valid` or `Invalid` statements on about these candidates.
issued_statements: HashSet,
/// `Some(h)` if this job has already issues `Seconded` statemt for some candidate with `h` hash.
seconded: Option,
/// We have already reported misbehaviors for these validators.
reported_misbehavior_for: HashSet,
table: Table,
table_context: TableContext,
}
const fn group_quorum(n_validators: usize) -> usize {
(n_validators / 2) + 1
}
#[derive(Default)]
struct TableContext {
signing_context: SigningContext,
validator: Option,
groups: HashMap>,
validators: Vec,
}
impl TableContextTrait for TableContext {
type AuthorityId = ValidatorIndex;
type Digest = Hash;
type GroupId = ParaId;
type Signature = ValidatorSignature;
type Candidate = CommittedCandidateReceipt;
fn candidate_digest(candidate: &CommittedCandidateReceipt) -> Hash {
candidate.hash()
}
fn candidate_group(candidate: &CommittedCandidateReceipt) -> ParaId {
candidate.descriptor().para_id
}
fn is_member_of(&self, authority: &ValidatorIndex, group: &ParaId) -> bool {
self.groups.get(group).map_or(false, |g| g.iter().position(|a| a == authority).is_some())
}
fn requisite_votes(&self, group: &ParaId) -> usize {
self.groups.get(group).map_or(usize::max_value(), |g| group_quorum(g.len()))
}
}
/// A message type that is sent from `CandidateBackingSubsystem` to `CandidateBackingJob`.
pub enum ToJob {
/// A `CandidateBackingMessage`.
CandidateBacking(CandidateBackingMessage),
/// Stop working.
Stop,
}
impl TryFrom for ToJob {
type Error = ();
fn try_from(msg: AllMessages) -> Result {
match msg {
AllMessages::CandidateBacking(msg) => Ok(ToJob::CandidateBacking(msg)),
_ => Err(()),
}
}
}
impl From for ToJob {
fn from(msg: CandidateBackingMessage) -> Self {
Self::CandidateBacking(msg)
}
}
impl util::ToJobTrait for ToJob {
const STOP: Self = ToJob::Stop;
fn relay_parent(&self) -> Option {
match self {
Self::CandidateBacking(cb) => cb.relay_parent(),
Self::Stop => None,
}
}
}
/// A message type that is sent from `CandidateBackingJob` to `CandidateBackingSubsystem`.
pub enum FromJob {
AvailabilityStore(AvailabilityStoreMessage),
RuntimeApiMessage(RuntimeApiMessage),
CandidateValidation(CandidateValidationMessage),
CandidateSelection(CandidateSelectionMessage),
Provisioner(ProvisionerMessage),
PoVDistribution(PoVDistributionMessage),
StatementDistribution(StatementDistributionMessage),
}
impl From for AllMessages {
fn from(f: FromJob) -> Self {
match f {
FromJob::AvailabilityStore(msg) => AllMessages::AvailabilityStore(msg),
FromJob::RuntimeApiMessage(msg) => AllMessages::RuntimeApi(msg),
FromJob::CandidateValidation(msg) => AllMessages::CandidateValidation(msg),
FromJob::CandidateSelection(msg) => AllMessages::CandidateSelection(msg),
FromJob::StatementDistribution(msg) => AllMessages::StatementDistribution(msg),
FromJob::PoVDistribution(msg) => AllMessages::PoVDistribution(msg),
FromJob::Provisioner(msg) => AllMessages::Provisioner(msg),
}
}
}
// It looks like it's not possible to do an `impl From` given the current state of
// the code. So this does the necessary conversion.
fn primitive_statement_to_table(s: &SignedFullStatement) -> TableSignedStatement {
let statement = match s.payload() {
Statement::Seconded(c) => TableStatement::Candidate(c.clone()),
Statement::Valid(h) => TableStatement::Valid(h.clone()),
Statement::Invalid(h) => TableStatement::Invalid(h.clone()),
};
TableSignedStatement {
statement,
signature: s.signature().clone(),
sender: s.validator_index(),
}
}
impl CandidateBackingJob {
/// Run asynchronously.
async fn run_loop(mut self) -> Result<(), Error> {
while let Some(msg) = self.rx_to.next().await {
match msg {
ToJob::CandidateBacking(msg) => {
self.process_msg(msg).await?;
}
_ => break,
}
}
Ok(())
}
async fn issue_candidate_invalid_message(
&mut self,
candidate: CandidateReceipt,
) -> Result<(), Error> {
self.tx_from.send(FromJob::CandidateSelection(
CandidateSelectionMessage::Invalid(self.parent, candidate)
)).await?;
Ok(())
}
/// Validate the candidate that is requested to be `Second`ed and distribute validation result.
///
/// Returns `Ok(true)` if we issued a `Seconded` statement about this candidate.
async fn validate_and_second(
&mut self,
candidate: &CandidateReceipt,
pov: PoV,
) -> Result {
let valid = self.request_candidate_validation(
candidate.descriptor().clone(),
Arc::new(pov.clone()),
).await?;
let candidate_hash = candidate.hash();
let statement = match valid {
ValidationResult::Valid(outputs) => {
// make PoV available for later distribution. Send data to the availability
// store to keep. Sign and dispatch `valid` statement to network if we
// have not seconded the given candidate.
//
// If the commitments hash produced by validation is not the same as given by
// the collator, do not make available and report the collator.
let commitments_check = self.make_pov_available(
pov,
outputs,
|commitments| if commitments.hash() == candidate.commitments_hash {
Ok(CommittedCandidateReceipt {
descriptor: candidate.descriptor().clone(),
commitments,
})
} else {
Err(())
},
).await?;
match commitments_check {
Ok(candidate) => {
self.issued_statements.insert(candidate_hash);
Some(Statement::Seconded(candidate))
}
Err(()) => {
self.issue_candidate_invalid_message(candidate.clone()).await?;
None
}
}
}
ValidationResult::Invalid => {
// no need to issue a statement about this if we aren't seconding it.
//
// there's an infinite amount of garbage out there. no need to acknowledge
// all of it.
self.issue_candidate_invalid_message(candidate.clone()).await?;
None
}
};
let issued_statement = statement.is_some();
if let Some(signed_statement) = statement.and_then(|s| self.sign_statement(s)) {
self.import_statement(&signed_statement).await?;
self.distribute_signed_statement(signed_statement).await?;
}
Ok(issued_statement)
}
fn get_backed(&self) -> Vec {
let proposed = self.table.proposed_candidates(&self.table_context);
let mut res = Vec::with_capacity(proposed.len());
for p in proposed.into_iter() {
let TableAttestedCandidate { candidate, validity_votes, .. } = p;
let (ids, validity_votes): (Vec<_>, Vec<_>) = validity_votes
.into_iter()
.map(|(id, vote)| (id, vote.into()))
.unzip();
let group = match self.table_context.groups.get(&self.assignment) {
Some(group) => group,
None => continue,
};
let mut validator_indices = BitVec::with_capacity(group.len());
validator_indices.resize(group.len(), false);
for id in ids.iter() {
if let Some(position) = group.iter().position(|x| x == id) {
validator_indices.set(position, true);
}
}
let backed = BackedCandidate {
candidate,
validity_votes,
validator_indices,
};
res.push(NewBackedCandidate(backed.clone()));
}
res
}
/// Check if there have happened any new misbehaviors and issue necessary messages.
///
/// TODO: Report multiple misbehaviors (https://github.com/paritytech/polkadot/issues/1387)
async fn issue_new_misbehaviors(&mut self) -> Result<(), Error> {
let mut reports = Vec::new();
for (k, v) in self.table.get_misbehavior().iter() {
if !self.reported_misbehavior_for.contains(k) {
self.reported_misbehavior_for.insert(*k);
let f = FromTableMisbehavior {
id: *k,
report: v.clone(),
signing_context: self.table_context.signing_context.clone(),
key: self.table_context.validators[*k as usize].clone(),
};
if let Ok(report) = MisbehaviorReport::try_from(f) {
let message = ProvisionerMessage::ProvisionableData(
ProvisionableData::MisbehaviorReport(self.parent, report),
);
reports.push(message);
}
}
}
for report in reports.drain(..) {
self.send_to_provisioner(report).await?
}
Ok(())
}
/// Import a statement into the statement table and return the summary of the import.
async fn import_statement(
&mut self,
statement: &SignedFullStatement,
) -> Result