pezkuwichain/pezkuwi-subxt
1
0
Fork 0
mirror of https://github.com/pezkuwichain/pezkuwi-subxt.git synced 2026-07-22 18:25:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

Wire up candidate backing, approval-voting to disputes (#3348)

Browse Source 
* add a from_backing_statement to SignedDisputeStatement

* inform dispute coordinator of all backing statements

* add dispute coordinator message to backing tests

* send positive dispute statement with every approval

* issue disputes when encountering invalid candidates.

* try to fix flaky test for CI (passed locally)

* guide: keep track of concluded-positive disputes until pruned

* guide: block implications

* guide: new dispute inherent flow

* mostly implement recency changes for dispute coordinator

* add a clock to dispute coordinator

* adjust DB tests

* fix and add new dispute coordinator tests

* provisioner: select disputes

* import all validators' approvals

* address nit: refactor backing statement submission

* gracefully handle disconnected dispute coordinator

* remove `review` comment

* fix up old_tests

* fix approval-voting compilation

* fix backing compilation

* use known-leaves in WaitForActivation

* follow-up test fixing

* add back allow(dead_code)
This commit is contained in:
Robert Habermeier
2021-07-09 16:15:51 -05:00
committed by GitHub
parent d53ec86bbe
commit e512222749
16 changed files with 1220 additions and 281 deletions
Download Patch File Download Diff File Expand all files Collapse all files
polkadot/node/core/backing/src/lib.rs
+106 -3
View File
@@ -30,9 +30,10 @@ use polkadot_primitives::v1::{
BackedCandidate, CandidateCommitments, CandidateDescriptor, CandidateHash,
CandidateReceipt, CollatorId, CommittedCandidateReceipt, CoreIndex, CoreState, Hash, Id as ParaId,
SigningContext, ValidatorId, ValidatorIndex, ValidatorSignature, ValidityAttestation,
SessionIndex,
};
use polkadot_node_primitives::{
Statement, SignedFullStatement, ValidationResult, PoV, AvailableData,
Statement, SignedFullStatement, ValidationResult, PoV, AvailableData, SignedDisputeStatement,
};
use polkadot_subsystem::{
PerLeafSpan, Stage, SubsystemSender,
@@ -42,7 +43,8 @@ use polkadot_subsystem::{
AllMessages, AvailabilityDistributionMessage, AvailabilityStoreMessage,
CandidateBackingMessage, CandidateValidationMessage, CollatorProtocolMessage,
ProvisionableData, ProvisionerMessage, RuntimeApiRequest,
StatementDistributionMessage, ValidationFailed
StatementDistributionMessage, ValidationFailed, DisputeCoordinatorMessage,
ImportStatementsResult,
}
};
use polkadot_node_subsystem_util::{
@@ -151,6 +153,8 @@ impl ValidatedCandidateCommand {
pub struct CandidateBackingJob {
/// The hash of the relay parent on top of which this job is doing it's work.
parent: Hash,
/// The session index this corresponds to.
session_index: SessionIndex,
/// The `ParaId` assigned to this validator
assignment: Option<ParaId>,
/// The collator required to author the candidate, if any.
@@ -538,6 +542,8 @@ async fn validate_and_make_available(
tx_command.send(make_command(res)).await.map_err(Into::into)
}
struct ValidatorIndexOutOfBounds;
impl CandidateBackingJob {
/// Run asynchronously.
async fn run_loop(
@@ -768,12 +774,28 @@ impl CandidateBackingJob {
"Importing statement",
);
let candidate_hash = statement.payload().candidate_hash();
let import_statement_span = {
// create a span only for candidates we're already aware of.
let candidate_hash = statement.payload().candidate_hash();
self.get_unbacked_statement_child(root_span, candidate_hash, statement.validator_index())
};
if let Err(ValidatorIndexOutOfBounds) = self.dispatch_new_statement_to_dispute_coordinator(
sender,
candidate_hash,
&statement,
).await {
tracing::warn!(
target: LOG_TARGET,
session_index = ?self.session_index,
relay_parent = ?self.parent,
validator_index = statement.validator_index().0,
"Supposedly 'Signed' statement has validator index out of bounds."
);
return Ok(None);
}
let stmt = primitive_statement_to_table(statement);
let summary = self.table.import_statement(&self.table_context, stmt);
@@ -824,6 +846,86 @@ impl CandidateBackingJob {
Ok(summary)
}
/// The dispute coordinator keeps track of all statements by validators about every recent
/// candidate.
///
/// When importing a statement, this should be called access the candidate receipt either
/// from the statement itself or from the underlying statement table in order to craft
/// and dispatch the notification to the dispute coordinator.
///
/// This also does bounds-checking on the validator index and will return an error if the
/// validator index is out of bounds for the current validator set. It's expected that
/// this should never happen due to the interface of the candidate backing subsystem -
/// the networking component repsonsible for feeding statements to the backing subsystem
/// is meant to check the signature and provenance of all statements before submission.
async fn dispatch_new_statement_to_dispute_coordinator(
&self,
sender: &mut JobSender<impl SubsystemSender>,
candidate_hash: CandidateHash,
statement: &SignedFullStatement,
) -> Result<(), ValidatorIndexOutOfBounds> {
// Dispatch the statement to the dispute coordinator.
let validator_index = statement.validator_index();
let signing_context = SigningContext {
parent_hash: self.parent,
session_index: self.session_index,
};
let validator_public = match self.table_context
.validators
.get(validator_index.0 as usize)
{
None => {
return Err(ValidatorIndexOutOfBounds);
}
Some(v) => v,
};
let maybe_candidate_receipt = match statement.payload() {
Statement::Seconded(receipt) => Some(receipt.to_plain()),
Statement::Valid(candidate_hash) => {
// Valid statements are only supposed to be imported
// once we've seen at least one `Seconded` statement.
self.table.get_candidate(&candidate_hash).map(|c| c.to_plain())
}
};
let maybe_signed_dispute_statement = SignedDisputeStatement::from_backing_statement(
statement.as_unchecked(),
signing_context,
validator_public.clone(),
).ok();
if let (Some(candidate_receipt), Some(dispute_statement))
= (maybe_candidate_receipt, maybe_signed_dispute_statement)
{
let (pending_confirmation, confirmation_rx) = oneshot::channel();
sender.send_message(
DisputeCoordinatorMessage::ImportStatements {
candidate_hash,
candidate_receipt,
session: self.session_index,
statements: vec![(dispute_statement, validator_index)],
pending_confirmation,
}
).await;
match confirmation_rx.await {
Err(oneshot::Canceled) => tracing::warn!(
target: LOG_TARGET,
"Dispute coordinator confirmation lost",
),
Ok(ImportStatementsResult::ValidImport) => {}
Ok(ImportStatementsResult::InvalidImport) => tracing::warn!(
target: LOG_TARGET,
"Failed to import statements of validity",
),
}
}
Ok(())
}
async fn process_msg(
&mut self,
root_span: &jaeger::Span,
@@ -1199,6 +1301,7 @@ impl util::JobTrait for CandidateBackingJob {
let (background_tx, background_rx) = mpsc::channel(16);
let job = CandidateBackingJob {
parent,
session_index,
assignment,
required_collator,
issued_statements: HashSet::new(),
polkadot/node/core/backing/src/tests.rs
+167 -23
View File
@@ -57,6 +57,12 @@ struct TestState {
relay_parent: Hash,
}
impl TestState {
fn session(&self) -> SessionIndex {
self.signing_context.session_index
}
}
impl Default for TestState {
fn default() -> Self {
let chain_a = ParaId::from(1);
@@ -259,6 +265,35 @@ async fn test_startup(
);
}
async fn test_dispute_coordinator_notifications(
virtual_overseer: &mut VirtualOverseer,
candidate_hash: CandidateHash,
session: SessionIndex,
validator_indices: Vec<ValidatorIndex>,
) {
for validator_index in validator_indices {
assert_matches!(
virtual_overseer.recv().await,
AllMessages::DisputeCoordinator(
DisputeCoordinatorMessage::ImportStatements {
candidate_hash: c_hash,
candidate_receipt: c_receipt,
session: s,
statements,
pending_confirmation,
}
) => {
assert_eq!(c_hash, candidate_hash);
assert_eq!(c_receipt.hash(), c_hash);
assert_eq!(s, session);
assert_eq!(statements.len(), 1);
assert_eq!(statements[0].1, validator_index);
let _ = pending_confirmation.send(ImportStatementsResult::ValidImport);
}
)
}
}
// Test that a `CandidateBackingMessage::Second` issues validation work
// and in case validation is successful issues a `StatementDistributionMessage`.
#[test]
@@ -291,7 +326,6 @@ fn backing_second_works() {
virtual_overseer.send(FromOverseer::Communication{ msg: second }).await;
assert_matches!(
virtual_overseer.recv().await,
AllMessages::CandidateValidation(
@@ -309,7 +343,7 @@ fn backing_second_works() {
new_validation_code: None,
processed_downward_messages: 0,
hrmp_watermark: 0,
}, test_state.validation_data),
}, test_state.validation_data.clone()),
)).unwrap();
}
);
@@ -323,6 +357,13 @@ fn backing_second_works() {
}
);
test_dispute_coordinator_notifications(
&mut virtual_overseer,
candidate.hash(),
test_state.session(),
vec![ValidatorIndex(0)],
).await;
assert_matches!(
virtual_overseer.recv().await,
AllMessages::StatementDistribution(
@@ -404,6 +445,13 @@ fn backing_works() {
virtual_overseer.send(FromOverseer::Communication{ msg: statement }).await;
test_dispute_coordinator_notifications(
&mut virtual_overseer,
candidate_a_hash,
test_state.session(),
vec![ValidatorIndex(2)],
).await;
// Sending a `Statement::Seconded` for our assignment will start
// validation process. The first thing requested is the PoV.
assert_matches!(
@@ -438,7 +486,7 @@ fn backing_works() {
new_validation_code: None,
processed_downward_messages: 0,
hrmp_watermark: 0,
}, test_state.validation_data),
}, test_state.validation_data.clone()),
)).unwrap();
}
);
@@ -452,6 +500,13 @@ fn backing_works() {
}
);
test_dispute_coordinator_notifications(
&mut virtual_overseer,
candidate_a_hash,
test_state.session(),
vec![ValidatorIndex(0)],
).await;
assert_matches!(
virtual_overseer.recv().await,
AllMessages::StatementDistribution(
@@ -468,6 +523,13 @@ fn backing_works() {
virtual_overseer.send(FromOverseer::Communication{ msg: statement }).await;
test_dispute_coordinator_notifications(
&mut virtual_overseer,
candidate_a_hash,
test_state.session(),
vec![ValidatorIndex(5)],
).await;
assert_matches!(
virtual_overseer.recv().await,
AllMessages::Provisioner(
@@ -554,6 +616,13 @@ fn backing_works_while_validation_ongoing() {
let statement = CandidateBackingMessage::Statement(test_state.relay_parent, signed_a.clone());
virtual_overseer.send(FromOverseer::Communication{ msg: statement }).await;
test_dispute_coordinator_notifications(
&mut virtual_overseer,
candidate_a.hash(),
test_state.session(),
vec![ValidatorIndex(2)],
).await;
// Sending a `Statement::Seconded` for our assignment will start
// validation process. The first thing requested is PoV from the
// `PoVDistribution`.
@@ -601,6 +670,13 @@ fn backing_works_while_validation_ongoing() {
virtual_overseer.send(FromOverseer::Communication{ msg: statement }).await;
test_dispute_coordinator_notifications(
&mut virtual_overseer,
candidate_a.hash(),
test_state.session(),
vec![ValidatorIndex(5), ValidatorIndex(3)],
).await;
// Candidate gets backed entirely by other votes.
assert_matches!(
virtual_overseer.recv().await,
@@ -699,6 +775,13 @@ fn backing_misbehavior_works() {
virtual_overseer.send(FromOverseer::Communication { msg: statement }).await;
test_dispute_coordinator_notifications(
&mut virtual_overseer,
candidate_a_hash,
test_state.session(),
vec![ValidatorIndex(2)],
).await;
assert_matches!(
virtual_overseer.recv().await,
AllMessages::AvailabilityDistribution(
@@ -729,7 +812,7 @@ fn backing_misbehavior_works() {
new_validation_code: None,
processed_downward_messages: 0,
hrmp_watermark: 0,
}, test_state.validation_data),
}, test_state.validation_data.clone()),
)).unwrap();
}
);
@@ -743,6 +826,13 @@ fn backing_misbehavior_works() {
}
);
test_dispute_coordinator_notifications(
&mut virtual_overseer,
candidate_a_hash,
test_state.session(),
vec![ValidatorIndex(0)],
).await;
assert_matches!(
virtual_overseer.recv().await,
AllMessages::StatementDistribution(
@@ -760,6 +850,13 @@ fn backing_misbehavior_works() {
virtual_overseer.send(FromOverseer::Communication { msg: statement }).await;
test_dispute_coordinator_notifications(
&mut virtual_overseer,
candidate_a_hash,
test_state.session(),
vec![ValidatorIndex(2)],
).await;
assert_matches!(
virtual_overseer.recv().await,
AllMessages::Provisioner(
@@ -889,7 +986,7 @@ fn backing_dont_second_invalid() {
new_validation_code: None,
processed_downward_messages: 0,
hrmp_watermark: 0,
}, test_state.validation_data),
}, test_state.validation_data.clone()),
)).unwrap();
}
);
@@ -903,6 +1000,13 @@ fn backing_dont_second_invalid() {
}
);
test_dispute_coordinator_notifications(
&mut virtual_overseer,
candidate_b.hash(),
test_state.session(),
vec![ValidatorIndex(0)],
).await;
assert_matches!(
virtual_overseer.recv().await,
AllMessages::StatementDistribution(
@@ -965,6 +1069,13 @@ fn backing_second_after_first_fails_works() {
virtual_overseer.send(FromOverseer::Communication{ msg: statement }).await;
test_dispute_coordinator_notifications(
&mut virtual_overseer,
candidate.hash(),
test_state.session(),
vec![ValidatorIndex(2)],
).await;
// Subsystem requests PoV and requests validation.
assert_matches!(
virtual_overseer.recv().await,
@@ -1087,6 +1198,13 @@ fn backing_works_after_failed_validation() {
virtual_overseer.send(FromOverseer::Communication{ msg: statement }).await;
test_dispute_coordinator_notifications(
&mut virtual_overseer,
candidate.hash(),
test_state.session(),
vec![ValidatorIndex(2)],
).await;
// Subsystem requests PoV and requests validation.
assert_matches!(
virtual_overseer.recv().await,
@@ -1375,6 +1493,13 @@ fn retry_works() {
);
virtual_overseer.send(FromOverseer::Communication{ msg: statement }).await;
test_dispute_coordinator_notifications(
&mut virtual_overseer,
candidate.hash(),
test_state.session(),
vec![ValidatorIndex(2)],
).await;
// Subsystem requests PoV and requests validation.
// We cancel - should mean retry on next backing statement.
assert_matches!(
@@ -1396,12 +1521,39 @@ fn retry_works() {
);
virtual_overseer.send(FromOverseer::Communication{ msg: statement }).await;
test_dispute_coordinator_notifications(
&mut virtual_overseer,
candidate.hash(),
test_state.session(),
vec![ValidatorIndex(3)],
).await;
assert_matches!(
virtual_overseer.recv().await,
AllMessages::AvailabilityDistribution(
AvailabilityDistributionMessage::FetchPoV {
relay_parent,
tx,
..
}
) if relay_parent == test_state.relay_parent => {
std::mem::drop(tx);
}
);
let statement = CandidateBackingMessage::Statement(
test_state.relay_parent,
signed_c.clone(),
);
virtual_overseer.send(FromOverseer::Communication{ msg: statement }).await;
test_dispute_coordinator_notifications(
&mut virtual_overseer,
candidate.hash(),
test_state.session(),
vec![ValidatorIndex(5)],
).await;
// Not deterministic which message comes first:
for _ in 0u32..2 {
match virtual_overseer.recv().await {
@@ -1417,15 +1569,15 @@ fn retry_works() {
assert_eq!(descriptor, candidate.descriptor);
}
// Subsystem requests PoV and requests validation.
// We cancel once more:
// Now we pass.
AllMessages::AvailabilityDistribution(
AvailabilityDistributionMessage::FetchPoV {
relay_parent,
tx,
..
}
) if relay_parent == test_state.relay_parent => {
std::mem::drop(tx);
) if relay_parent == test_state.relay_parent => {
tx.send(pov.clone()).unwrap();
}
msg => {
assert!(false, "Unexpected message: {:?}", msg);
@@ -1433,21 +1585,6 @@ fn retry_works() {
}
}
// Subsystem requests PoV and requests validation.
// Now we pass.
assert_matches!(
virtual_overseer.recv().await,
AllMessages::AvailabilityDistribution(
AvailabilityDistributionMessage::FetchPoV {
relay_parent,
tx,
..
}
) if relay_parent == test_state.relay_parent => {
tx.send(pov.clone()).unwrap();
}
);
assert_matches!(
virtual_overseer.recv().await,
AllMessages::CandidateValidation(
@@ -1547,6 +1684,13 @@ fn observes_backing_even_if_not_validator() {
virtual_overseer.send(FromOverseer::Communication{ msg: statement }).await;
test_dispute_coordinator_notifications(
&mut virtual_overseer,
candidate_a_hash,
test_state.session(),
vec![ValidatorIndex(0), ValidatorIndex(5), ValidatorIndex(2)],
).await;
assert_matches!(
virtual_overseer.recv().await,
AllMessages::Provisioner(