Unverified Commit ed759c75 authored by Sergey Pepyakin's avatar Sergey Pepyakin Committed by GitHub
Browse files

Downward Message Processing implementation (#1859)

* DMP: data structures and plumbing

* DMP: Implement DMP logic in the router module

DMP: Integrate DMP parts into the inclusion module

* DMP: Introduce the max size limit for the size of a downward message

* DMP: Runtime API for accessing inbound messages

* OCD

Small clean ups

* DMP: fix the naming of the error

* DMP: add caution about a non-existent recipient
parent 798f781f
Pipeline #112381 failed with stages
in 9 minutes and 18 seconds
......@@ -79,20 +79,19 @@ pub use sp_runtime::OpaqueExtrinsic as UncheckedExtrinsic;
/// the parachain.
pub type Remark = [u8; 32];
/// These are special "control" messages that can be passed from the Relaychain to a parachain.
/// They should be handled by all parachains.
/// A message sent from the relay-chain down to a parachain.
///
/// The size of the message is limited by the `config.max_downward_message_size` parameter.
pub type DownwardMessage = sp_std::vec::Vec<u8>;
/// A wrapped version of `DownwardMessage`. The difference is that it has attached the block number when
/// the message was sent.
#[derive(codec::Encode, codec::Decode, Clone, sp_runtime::RuntimeDebug, PartialEq)]
pub enum DownwardMessage<AccountId = crate::AccountId> {
/// Some funds were transferred into the parachain's account. The hash is the identifier that
/// was given with the transfer.
TransferInto(AccountId, Balance, Remark),
/// An opaque blob of data. The relay chain must somehow know how to form this so that the
/// destination parachain does something sensible.
///
/// NOTE: Be very careful not to allow users to place arbitrary size information in here.
Opaque(sp_std::vec::Vec<u8>),
/// XCMP message for the Parachain.
XCMPMessage(sp_std::vec::Vec<u8>),
pub struct InboundDownwardMessage<BlockNumber = crate::BlockNumber> {
/// The block number at which this messages was put into the downward message queue.
pub sent_at: BlockNumber,
/// The actual downward message to processes.
pub msg: DownwardMessage,
}
/// V1 primitives.
......
......@@ -277,6 +277,7 @@ async fn handle_new_activations<Context: SubsystemContext>(
new_validation_code: collation.new_validation_code,
head_data: collation.head_data,
erasure_root,
processed_downward_messages: collation.processed_downward_messages,
};
let ccr = CandidateReceipt {
......@@ -387,6 +388,7 @@ mod tests {
proof_of_validity: PoV {
block_data: BlockData(Vec::new()),
},
processed_downward_messages: Default::default(),
}
}
......
......@@ -70,6 +70,7 @@ impl Default for TestState {
parent_head: HeadData(vec![7, 8, 9]),
block_number: 5,
hrmp_mqc_heads: Vec::new(),
dmq_mqc_head: Default::default(),
};
let pruning_config = PruningConfig {
......
......@@ -682,6 +682,7 @@ impl CandidateBackingJob {
erasure_root,
new_validation_code: outputs.new_validation_code,
head_data: outputs.head_data,
processed_downward_messages: outputs.processed_downward_messages,
};
let res = match with_commitments(commitments) {
......@@ -977,12 +978,14 @@ mod tests {
parent_head: HeadData(vec![7, 8, 9]),
block_number: Default::default(),
hrmp_mqc_heads: Vec::new(),
dmq_mqc_head: Default::default(),
},
transient: TransientValidationData {
max_code_size: 1000,
max_head_data_size: 1000,
balance: Default::default(),
code_upgrade_allowed: None,
dmq_length: 0,
},
};
......@@ -1159,6 +1162,7 @@ mod tests {
upward_messages: Vec::new(),
fees: Default::default(),
new_validation_code: None,
processed_downward_messages: 0,
}, test_state.validation_data.persisted),
)).unwrap();
}
......@@ -1278,6 +1282,7 @@ mod tests {
upward_messages: Vec::new(),
fees: Default::default(),
new_validation_code: None,
processed_downward_messages: 0,
}, test_state.validation_data.persisted),
)).unwrap();
}
......@@ -1416,6 +1421,7 @@ mod tests {
upward_messages: Vec::new(),
fees: Default::default(),
new_validation_code: None,
processed_downward_messages: 0,
}, test_state.validation_data.persisted),
)).unwrap();
}
......@@ -1571,6 +1577,7 @@ mod tests {
upward_messages: Vec::new(),
fees: Default::default(),
new_validation_code: None,
processed_downward_messages: 0,
}, test_state.validation_data.persisted),
)).unwrap();
}
......
......@@ -494,11 +494,13 @@ mod tests {
upward_messages: Vec::new(),
fees: 0,
new_validation_code: None,
processed_downward_messages: 0,
},
PersistedValidationData {
parent_head: HeadData(parent_head_data),
block_number: 123,
hrmp_mqc_heads: Vec::new(),
dmq_mqc_head: Default::default(),
},
)
}
......
......@@ -468,6 +468,7 @@ fn validate_candidate_exhaustive<B: ValidationBackend, S: SpawnNamed + 'static>(
parent_head: persisted_validation_data.parent_head.clone(),
block_data: pov.block_data.clone(),
relay_chain_height: persisted_validation_data.block_number,
dmq_mqc_head: persisted_validation_data.dmq_mqc_head,
hrmp_mqc_heads: persisted_validation_data.hrmp_mqc_heads.clone(),
};
......@@ -491,6 +492,7 @@ fn validate_candidate_exhaustive<B: ValidationBackend, S: SpawnNamed + 'static>(
upward_messages: res.upward_messages,
fees: 0,
new_validation_code: res.new_validation_code,
processed_downward_messages: res.processed_downward_messages,
};
Ok(ValidationResult::Valid(outputs, persisted_validation_data))
}
......
......@@ -127,6 +127,7 @@ fn make_runtime_api_request<Client>(
query!(candidate_pending_availability(para), sender),
Request::CandidateEvents(sender) => query!(candidate_events(), sender),
Request::ValidatorDiscovery(ids, sender) => query!(validator_discovery(ids), sender),
Request::DmqContents(id, sender) => query!(dmq_contents(id), sender),
}
}
......@@ -176,7 +177,7 @@ mod tests {
use polkadot_primitives::v1::{
ValidatorId, ValidatorIndex, GroupRotationInfo, CoreState, PersistedValidationData,
Id as ParaId, OccupiedCoreAssumption, ValidationData, SessionIndex, ValidationCode,
CommittedCandidateReceipt, CandidateEvent, AuthorityDiscoveryId,
CommittedCandidateReceipt, CandidateEvent, AuthorityDiscoveryId, InboundDownwardMessage,
};
use polkadot_node_subsystem_test_helpers as test_helpers;
use sp_core::testing::TaskExecutor;
......@@ -195,6 +196,7 @@ mod tests {
validation_outputs_results: HashMap<ParaId, bool>,
candidate_pending_availability: HashMap<ParaId, CommittedCandidateReceipt>,
candidate_events: Vec<CandidateEvent>,
dmq_contents: HashMap<ParaId, Vec<InboundDownwardMessage>>,
}
impl ProvideRuntimeApi<Block> for MockRuntimeApi {
......@@ -283,6 +285,13 @@ mod tests {
fn validator_discovery(ids: Vec<ValidatorId>) -> Vec<Option<AuthorityDiscoveryId>> {
vec![None; ids.len()]
}
fn dmq_contents(
&self,
recipient: ParaId,
) -> Vec<polkadot_primitives::v1::InboundDownwardMessage> {
self.dmq_contents.get(&recipient).map(|q| q.clone()).unwrap_or_default()
}
}
}
......@@ -619,4 +628,54 @@ mod tests {
futures::executor::block_on(future::join(subsystem_task, test_task));
}
#[test]
fn requests_dmq_contents() {
let (ctx, mut ctx_handle) = test_helpers::make_subsystem_context(TaskExecutor::new());
let mut runtime_api = MockRuntimeApi::default();
let relay_parent = [1; 32].into();
let para_a = 5.into();
let para_b = 6.into();
runtime_api.dmq_contents.insert(para_a, vec![]);
runtime_api.dmq_contents.insert(
para_b,
vec![InboundDownwardMessage {
sent_at: 228,
msg: b"Novus Ordo Seclorum".to_vec(),
}],
);
let subsystem = RuntimeApiSubsystem::new(runtime_api.clone(), Metrics(None));
let subsystem_task = run(ctx, subsystem).map(|x| x.unwrap());
let test_task = async move {
let (tx, rx) = oneshot::channel();
ctx_handle
.send(FromOverseer::Communication {
msg: RuntimeApiMessage::Request(relay_parent, Request::DmqContents(para_a, tx)),
})
.await;
assert_eq!(rx.await.unwrap().unwrap(), vec![]);
let (tx, rx) = oneshot::channel();
ctx_handle
.send(FromOverseer::Communication {
msg: RuntimeApiMessage::Request(relay_parent, Request::DmqContents(para_b, tx)),
})
.await;
assert_eq!(
rx.await.unwrap().unwrap(),
vec![InboundDownwardMessage {
sent_at: 228,
msg: b"Novus Ordo Seclorum".to_vec(),
}]
);
ctx_handle
.send(FromOverseer::Signal(OverseerSignal::Conclude))
.await;
};
futures::executor::block_on(future::join(subsystem_task, test_task));
}
}
......@@ -217,6 +217,7 @@ impl Default for TestState {
parent_head: HeadData(vec![7, 8, 9]),
block_number: Default::default(),
hrmp_mqc_heads: Vec::new(),
dmq_mqc_head: Default::default(),
};
let validator_index = Some((validators.len() - 1) as ValidatorIndex);
......
......@@ -263,6 +263,8 @@ pub struct Collation {
pub head_data: HeadData,
/// Proof to verify the state transition of the parachain.
pub proof_of_validity: PoV,
/// The number of messages processed from the DMQ.
pub processed_downward_messages: u32,
}
/// Configuration for the collation generator
......
......@@ -37,7 +37,7 @@ use polkadot_primitives::v1::{
GroupRotationInfo, Hash, Id as ParaId, OccupiedCoreAssumption,
PersistedValidationData, PoV, SessionIndex, SignedAvailabilityBitfield,
ValidationCode, ValidatorId, ValidationData,
ValidatorIndex, ValidatorSignature,
ValidatorIndex, ValidatorSignature, InboundDownwardMessage,
};
use std::sync::Arc;
......@@ -419,7 +419,11 @@ pub enum RuntimeApiRequest {
/// Get the validation code for a para, taking the given `OccupiedCoreAssumption`, which
/// will inform on how the validation data should be computed if the para currently
/// occupies a core.
ValidationCode(ParaId, OccupiedCoreAssumption, RuntimeApiSender<Option<ValidationCode>>),
ValidationCode(
ParaId,
OccupiedCoreAssumption,
RuntimeApiSender<Option<ValidationCode>>,
),
/// Get a the candidate pending availability for a particular parachain by parachain / core index
CandidatePendingAvailability(ParaId, RuntimeApiSender<Option<CommittedCandidateReceipt>>),
/// Get all events concerning candidates (backing, inclusion, time-out) in the parent of
......@@ -429,7 +433,15 @@ pub enum RuntimeApiRequest {
/// Currently this request is limited to validators in the current session.
///
/// Returns `None` for validators not found in the current session.
ValidatorDiscovery(Vec<ValidatorId>, RuntimeApiSender<Vec<Option<AuthorityDiscoveryId>>>),
ValidatorDiscovery(
Vec<ValidatorId>,
RuntimeApiSender<Vec<Option<AuthorityDiscoveryId>>>,
),
/// Get all the pending inbound messages in the downward message queue for a para.
DmqContents(
ParaId,
RuntimeApiSender<Vec<InboundDownwardMessage<BlockNumber>>>,
),
}
/// A message to the Runtime API subsystem.
......
......@@ -248,6 +248,11 @@ pub struct ValidationParams {
pub block_data: BlockData,
/// The current relay-chain block number.
pub relay_chain_height: RelayChainBlockNumber,
/// The MQC head for the DMQ.
///
/// The DMQ MQC head will be used by the validation function to authorize the downward messages
/// passed by the collator.
pub dmq_mqc_head: Hash,
/// The list of MQC heads for the inbound HRMP channels paired with the sender para ids. This
/// vector is sorted ascending by the para id and doesn't contain multiple entries with the same
/// sender.
......
......@@ -97,6 +97,7 @@ fn execute_good_on_parent(execution_mode: ExecutionMode) {
block_data: GenericBlockData(block_data.encode()),
relay_chain_height: 1,
hrmp_mqc_heads: Vec::new(),
dmq_mqc_head: Default::default(),
},
&execution_mode,
sp_core::testing::TaskExecutor::new(),
......@@ -135,6 +136,7 @@ fn execute_good_chain_on_parent() {
block_data: GenericBlockData(block_data.encode()),
relay_chain_height: number as RelayChainBlockNumber + 1,
hrmp_mqc_heads: Vec::new(),
dmq_mqc_head: Default::default(),
},
&execution_mode,
sp_core::testing::TaskExecutor::new(),
......@@ -174,6 +176,7 @@ fn execute_bad_on_parent() {
block_data: GenericBlockData(block_data.encode()),
relay_chain_height: 1,
hrmp_mqc_heads: Vec::new(),
dmq_mqc_head: Default::default(),
},
&execution_mode,
sp_core::testing::TaskExecutor::new(),
......
......@@ -43,6 +43,7 @@ fn terminates_on_timeout() {
parent_head: Default::default(),
relay_chain_height: 1,
hrmp_mqc_heads: Vec::new(),
dmq_mqc_head: Default::default(),
},
&execution_mode,
sp_core::testing::TaskExecutor::new(),
......@@ -71,6 +72,7 @@ fn parallel_execution() {
parent_head: Default::default(),
relay_chain_height: 1,
hrmp_mqc_heads: Vec::new(),
dmq_mqc_head: Default::default(),
},
&execution_mode,
sp_core::testing::TaskExecutor::new(),
......@@ -82,6 +84,7 @@ fn parallel_execution() {
parent_head: Default::default(),
relay_chain_height: 1,
hrmp_mqc_heads: Vec::new(),
dmq_mqc_head: Default::default(),
},
&execution_mode2,
sp_core::testing::TaskExecutor::new(),
......
......@@ -31,7 +31,7 @@ pub use runtime_primitives::traits::{BlakeTwo256, Hash as HashT};
pub use polkadot_core_primitives::v1::{
BlockNumber, Moment, Signature, AccountPublic, AccountId, AccountIndex,
ChainId, Hash, Nonce, Balance, Header, Block, BlockId, UncheckedExtrinsic,
Remark, DownwardMessage,
Remark, DownwardMessage, InboundDownwardMessage,
};
// Export some polkadot-parachain primitives
......@@ -266,6 +266,11 @@ pub struct PersistedValidationData<N = BlockNumber> {
/// vector is sorted ascending by the para id and doesn't contain multiple entries with the same
/// sender.
pub hrmp_mqc_heads: Vec<(Id, Hash)>,
/// The MQC head for the DMQ.
///
/// The DMQ MQC head will be used by the validation function to authorize the downward messages
/// passed by the collator.
pub dmq_mqc_head: Hash,
}
impl<N: Encode> PersistedValidationData<N> {
......@@ -301,6 +306,8 @@ pub struct TransientValidationData<N = BlockNumber> {
/// which case the code upgrade should be applied at the end of the signaling
/// block.
pub code_upgrade_allowed: Option<N>,
/// The number of messages pending of the downward message queue.
pub dmq_length: u32,
}
/// Outputs of validating a candidate.
......@@ -315,6 +322,8 @@ pub struct ValidationOutputs {
pub fees: Balance,
/// The new validation code submitted by the execution, if any.
pub new_validation_code: Option<ValidationCode>,
/// The number of messages processed from the DMQ.
pub processed_downward_messages: u32,
}
/// Commitments made in a `CandidateReceipt`. Many of these are outputs of validation.
......@@ -331,6 +340,8 @@ pub struct CandidateCommitments {
pub new_validation_code: Option<ValidationCode>,
/// The head-data produced as a result of execution.
pub head_data: HeadData,
/// The number of messages processed from the DMQ.
pub processed_downward_messages: u32,
}
impl CandidateCommitments {
......@@ -712,6 +723,11 @@ sp_api::decl_runtime_apis! {
/// We assume that every validator runs authority discovery,
/// which would allow us to establish point-to-point connection to given validators.
fn validator_discovery(validators: Vec<ValidatorId>) -> Vec<Option<AuthorityDiscoveryId>>;
/// Get all the pending inbound messages in the downward message queue for a para.
fn dmq_contents(
recipient: Id,
) -> Vec<InboundDownwardMessage<N>>;
}
}
......
......@@ -18,7 +18,7 @@
#![cfg_attr(not(feature = "std"), no_std)]
// `construct_runtime!` does a lot of recursion and requires us to increase the limit to 256.
#![recursion_limit="256"]
#![recursion_limit = "256"]
use sp_std::prelude::*;
use sp_core::u32_trait::{_1, _2, _3, _4, _5};
......@@ -1134,6 +1134,12 @@ sp_api::impl_runtime_apis! {
fn validator_discovery(_: Vec<ValidatorId>) -> Vec<Option<AuthorityDiscoveryId>> {
Vec::new()
}
fn dmq_contents(
_recipient: Id,
) -> Vec<primitives::v1::InboundDownwardMessage<BlockNumber>> {
Vec::new()
}
}
impl fg_primitives::GrandpaApi<Block> for Runtime {
......
......@@ -58,6 +58,13 @@ pub struct HostConfiguration<BlockNumber> {
pub thread_availability_period: BlockNumber,
/// The amount of blocks ahead to schedule parachains and parathreads.
pub scheduling_lookahead: u32,
/// The maximum size of a message that can be put in a downward message queue.
///
/// Since we require receiving at least one DMP message the obvious upper bound of the size is
/// the PoV size. Of course, there is a lot of other different things that a parachain may
/// decide to do with its PoV so this value in practice will be picked as a fraction of the PoV
/// size.
pub max_downward_message_size: u32,
}
pub trait Trait: frame_system::Trait { }
......@@ -190,6 +197,16 @@ decl_module! {
});
Ok(())
}
/// Set the critical downward message size.
#[weight = (1_000, DispatchClass::Operational)]
pub fn set_max_downward_message_size(origin, new: u32) -> DispatchResult {
ensure_root(origin)?;
Self::update_config_member(|config| {
sp_std::mem::replace(&mut config.max_downward_message_size, new) != new
});
Ok(())
}
}
}
......@@ -268,6 +285,7 @@ mod tests {
chain_availability_period: 10,
thread_availability_period: 8,
scheduling_lookahead: 3,
max_downward_message_size: 2048,
};
assert!(<Configuration as Store>::PendingConfig::get().is_none());
......@@ -305,6 +323,9 @@ mod tests {
Configuration::set_scheduling_lookahead(
Origin::root(), new_config.scheduling_lookahead,
).unwrap();
Configuration::set_max_downward_message_size(
Origin::root(), new_config.max_downward_message_size,
).unwrap();
assert_eq!(<Configuration as Store>::PendingConfig::get(), Some(new_config));
})
......
......@@ -36,7 +36,7 @@ use bitvec::{order::Lsb0 as BitOrderLsb0, vec::BitVec};
use sp_staking::SessionIndex;
use sp_runtime::{DispatchError, traits::{One, Saturating}};
use crate::{configuration, paras, scheduler::CoreAssignment};
use crate::{configuration, paras, router, scheduler::CoreAssignment};
/// A bitfield signed by a validator indicating that it is keeping its piece of the erasure-coding
/// for any backed candidates referred to by a `1` bit available.
......@@ -86,7 +86,7 @@ impl<H, N> CandidatePendingAvailability<H, N> {
}
pub trait Trait:
frame_system::Trait + paras::Trait + configuration::Trait
frame_system::Trait + paras::Trait + router::Trait + configuration::Trait
{
type Event: From<Event<Self>> + Into<<Self as frame_system::Trait>::Event>;
}
......@@ -153,6 +153,8 @@ decl_error! {
ValidationDataHashMismatch,
/// Internal error only returned when compiled with debug assertions.
InternalError,
/// The downward message queue is not processed correctly.
IncorrectDownwardMessageHandling,
}
}
......@@ -409,6 +411,7 @@ impl<T: Trait> Module<T> {
para_id,
&candidate.candidate.commitments.head_data,
&candidate.candidate.commitments.new_validation_code,
candidate.candidate.commitments.processed_downward_messages,
)?;
for (i, assignment) in scheduled[skip..].iter().enumerate() {
......@@ -540,6 +543,7 @@ impl<T: Trait> Module<T> {
para_id,
&validation_outputs.head_data,
&validation_outputs.new_validation_code,
validation_outputs.processed_downward_messages,
)
}
......@@ -561,6 +565,12 @@ impl<T: Trait> Module<T> {
);
}
// enact the messaging facet of the candidate.
weight += <router::Module<T>>::prune_dmq(
receipt.descriptor.para_id,
commitments.processed_downward_messages,
);
Self::deposit_event(
Event::<T>::CandidateIncluded(plain, commitments.head_data.clone())
);
......@@ -682,6 +692,7 @@ impl<T: Trait> CandidateCheckContext<T> {
para_id: ParaId,
head_data: &HeadData,
new_validation_code: &Option<primitives::v1::ValidationCode>,
processed_downward_messages: u32,
) -> Result<(), DispatchError> {
ensure!(
head_data.0.len() <= self.config.max_head_data_size as _,
......@@ -703,7 +714,14 @@ impl<T: Trait> CandidateCheckContext<T> {
);
}
// TODO: messaging acceptance criteria rules will go here.
// check if the candidate passes the messaging acceptance criteria
ensure!(
<router::Module<T>>::check_processed_downward_messages(
para_id,
processed_downward_messages,
),
Error::<T>::IncorrectDownwardMessageHandling,
);
Ok(())
}
......
......@@ -128,8 +128,6 @@ pub type Configuration = crate::configuration::Module<Test>;
pub type Paras = crate::paras::Module<Test>;
/// Mocked router.
// TODO: Will be used in the follow ups.
#[allow(dead_code)]
pub type Router = crate::router::Module<Test>;
/// Mocked scheduler.
......
......@@ -20,10 +20,17 @@
//! routing the messages at their destinations and informing the parachains about the incoming
//! messages.
use crate::{configuration, initializer};
use crate::{
configuration,
initializer,
};
use sp_std::prelude::*;
use frame_support::{decl_error, decl_module, decl_storage, weights::Weight};
use primitives::v1::{Id as ParaId};
use primitives::v1::{Id as ParaId, InboundDownwardMessage, Hash};
mod dmp;
pub use dmp::QueueDownwardMessageError;
pub trait Trait: frame_system::Trait + configuration::Trait {}
......@@ -32,6 +39,23 @@ decl_storage! {
/// Paras that are to be cleaned up at the end of the session.
/// The entries are sorted ascending by the para id.
OutgoingParas: Vec<ParaId>;
/*
* Downward Message Passing (DMP)
*
* Storage layout required for implementation of DMP.
*/
/// The downward messages addressed for a certain para.
DownwardMessageQueues: map hasher(twox_64_concat) ParaId => Vec<InboundDownwardMessage<T::BlockNumber>>;
/// A mapping that stores the downward message queue MQC head for each para.
///
/// Each link in this chain has a form:
/// `(prev_head, B, H(M))`, where
/// - `prev_head`: is the previous head hash or zero if none.
/// - `B`: is the relay-chain block number in which a message was appended.
/// - `H(M)`: is the hash of the message being appended.
DownwardMessageQueueHeads: map hasher(twox_64_concat) ParaId => Hash;
}
}
......@@ -60,8 +84,8 @@ impl<T: Trait> Module<T> {