pezkuwichain/pezkuwi-subxt
1
0
Fork 0
mirror of https://github.com/pezkuwichain/pezkuwi-subxt.git synced 2026-07-21 17:55:42 +00:00
Code Issues Packages Projects Releases Wiki Activity

Migrate from MQCs in persisted validation data to merkle proofs (#317)

Browse Source 
* Update polkadot

* Migrate all uses of MQC heads to merkle proofs

* Mass rename `relay_parent_storage_root`

* Restore parachain-system tests

* Update polkadot and libp2p swarm for testing

* Collapse match into an if let

Co-authored-by: Bastian Köcher <bkchr@users.noreply.github.com>

* Fix compilation error in test-service

Co-authored-by: Bastian Köcher <bkchr@users.noreply.github.com>
This commit is contained in:
Sergei Shulepov
2021-02-08 19:48:30 +01:00
committed by GitHub
parent b32ce07176
commit e065c5776b
10 changed files with 553 additions and 398 deletions
Download Patch File Download Diff File Expand all files Collapse all files
parachain-system/src/lib.rs
+62 -72
View File
@@ -179,18 +179,18 @@ decl_module! {
// which means we can put the initialization logic here to remove the
// sequencing problem.
if let Some((apply_block, validation_function)) = PendingValidationFunction::get() {
if vfp.block_number >= apply_block {
if vfp.relay_parent_number >= apply_block {
PendingValidationFunction::kill();
LastUpgrade::put(&apply_block);
Self::put_parachain_code(&validation_function);
Self::deposit_event(Event::ValidationFunctionApplied(vfp.block_number));
Self::deposit_event(Event::ValidationFunctionApplied(vfp.relay_parent_number));
}
}
let (host_config, relevant_messaging_state) =
relay_state_snapshot::extract_from_proof(
T::SelfParaId::get(),
vfp.relay_storage_root,
vfp.relay_parent_storage_root,
relay_chain_state
)
.map_err(|err| {
@@ -200,13 +200,19 @@ decl_module! {
storage::unhashed::put(VALIDATION_DATA, &vfp);
DidUpdateValidationData::put(true);
RelevantMessagingState::put(relevant_messaging_state);
RelevantMessagingState::put(relevant_messaging_state.clone());
HostConfiguration::put(host_config);
<T::OnValidationData as OnValidationData>::on_validation_data(&vfp);
Self::process_inbound_downward_messages(&vfp, downward_messages)?;
Self::process_inbound_horizontal_messages(&vfp, horizontal_messages)?;
Self::process_inbound_downward_messages(
relevant_messaging_state.dmq_mqc_head,
downward_messages,
)?;
Self::process_inbound_horizontal_messages(
&relevant_messaging_state.ingress_channels,
horizontal_messages,
)?;
Ok(())
}
@@ -437,7 +443,7 @@ impl<T: Config> Module<T> {
/// Checks if the sequence of the messages is valid, dispatches them and communicates the number
/// of processed messages to the collator via a storage update.
fn process_inbound_downward_messages(
vfp: &PersistedValidationData,
expected_dmq_mqc_head: relay_chain::Hash,
downward_messages: Vec<InboundDownwardMessage>,
) -> DispatchResult {
let dm_count = downward_messages.len() as u32;
@@ -453,7 +459,7 @@ impl<T: Config> Module<T> {
// After hashing each message in the message queue chain submitted by the collator, we should
// arrive to the MQC head provided by the relay chain.
ensure!(
result_mqc_head == vfp.dmq_mqc_head,
result_mqc_head == expected_dmq_mqc_head,
Error::<T>::DmpMqcMismatch
);
@@ -469,14 +475,14 @@ impl<T: Config> Module<T> {
/// This is similar to [`process_inbound_downward_messages`], but works on multiple inbound
/// channels.
fn process_inbound_horizontal_messages(
vfp: &PersistedValidationData,
ingress_channels: &[(ParaId, cumulus_primitives::AbridgedHrmpChannel)],
horizontal_messages: BTreeMap<ParaId, Vec<InboundHrmpMessage>>,
) -> DispatchResult {
// First, check that all submitted messages are sent from channels that exist. The channel
// exists if its MQC head is present in `vfp.hrmp_mqc_heads`.
for sender in horizontal_messages.keys() {
ensure!(
vfp.hrmp_mqc_heads
ingress_channels
.binary_search_by_key(sender, |&(s, _)| s)
.is_ok(),
Error::<T>::HrmpNoMqc,
@@ -533,13 +539,14 @@ impl<T: Config> Module<T> {
// `running_mqc_heads`. Otherwise, in a block where no messages were sent in a channel
// it won't get into next block's `last_mqc_heads` and thus will be all zeros, which
// would corrupt the message queue chain.
for &(ref sender, ref target_head) in &vfp.hrmp_mqc_heads {
for &(ref sender, ref channel) in ingress_channels {
let cur_head = running_mqc_heads
.entry(*sender)
.or_insert_with(|| last_mqc_heads.get(&sender).cloned().unwrap_or_default())
.head();
let target_head = channel.mqc_head.unwrap_or_default();
ensure!(&cur_head == target_head, Error::<T>::HrmpMqcMismatch);
ensure!(cur_head == target_head, Error::<T>::HrmpMqcMismatch);
}
LastHrmpMqcHeads::put(running_mqc_heads);
@@ -592,7 +599,7 @@ impl<T: Config> Module<T> {
}
let relay_blocks_since_last_upgrade = vfp
.block_number
.relay_parent_number
.saturating_sub(LastUpgrade::get());
if relay_blocks_since_last_upgrade <= cfg.validation_upgrade_frequency {
@@ -600,7 +607,7 @@ impl<T: Config> Module<T> {
return None;
}
Some(vfp.block_number + cfg.validation_upgrade_delay)
Some(vfp.relay_parent_number + cfg.validation_upgrade_delay)
}
/// The implementation of the runtime upgrade scheduling.
@@ -1072,6 +1079,7 @@ mod tests {
self
}
#[allow(dead_code)] // might come in handy in future. If now is future and it still hasn't - feel free.
fn with_validation_data<F>(mut self, f: F) -> Self
where
F: 'static + Fn(&BlockTests, RelayChainBlockNumber, &mut PersistedValidationData),
@@ -1117,11 +1125,11 @@ mod tests {
if let Some(ref hook) = self.relay_sproof_builder_hook {
hook(self, *n as RelayChainBlockNumber, &mut sproof_builder);
}
let (relay_storage_root, relay_chain_state) =
let (relay_parent_storage_root, relay_chain_state) =
sproof_builder.into_state_root_and_proof();
let mut vfp = PersistedValidationData {
block_number: *n as RelayChainBlockNumber,
relay_storage_root,
relay_parent_number: *n as RelayChainBlockNumber,
relay_parent_storage_root,
..Default::default()
};
if let Some(ref hook) = self.persisted_validation_data_hook {
@@ -1612,11 +1620,11 @@ mod tests {
}
BlockTests::new()
.with_validation_data(
|_, relay_block_num, validation_data| match relay_block_num {
.with_relay_sproof_builder(
|_, relay_block_num, sproof| match relay_block_num {
1 => {
validation_data.dmq_mqc_head =
MessageQueueChain::default().extend_downward(&MSG).head();
sproof.dmq_mqc_head =
Some(MessageQueueChain::default().extend_downward(&MSG).head());
}
_ => unreachable!(),
},
@@ -1661,39 +1669,31 @@ mod tests {
}
BlockTests::new()
.with_validation_data(
|_, relay_block_num, validation_data| match relay_block_num {
.with_relay_sproof_builder(
|_, relay_block_num, sproof| match relay_block_num {
1 => {
// 200 - doesn't exist yet
// 300 - one new message
validation_data.hrmp_mqc_heads.push((
ParaId::from(300),
MessageQueueChain::default().extend_hrmp(&MSG_1).head(),
));
sproof.upsert_inbound_channel(ParaId::from(300)).mqc_head =
Some(MessageQueueChain::default().extend_hrmp(&MSG_1).head());
}
2 => {
// 200 - two new messages
// 300 - now present with one message.
validation_data.hrmp_mqc_heads.push((
ParaId::from(200),
MessageQueueChain::default().extend_hrmp(&MSG_4).head(),
));
validation_data.hrmp_mqc_heads.push((
ParaId::from(300),
MessageQueueChain::default()
sproof.upsert_inbound_channel(ParaId::from(200)).mqc_head =
Some(MessageQueueChain::default().extend_hrmp(&MSG_4).head());
sproof.upsert_inbound_channel(ParaId::from(300)).mqc_head =
Some(MessageQueueChain::default()
.extend_hrmp(&MSG_1)
.extend_hrmp(&MSG_2)
.extend_hrmp(&MSG_3)
.head(),
));
.head());
}
3 => {
// 200 - no new messages
// 300 - is gone
validation_data.hrmp_mqc_heads.push((
ParaId::from(200),
MessageQueueChain::default().extend_hrmp(&MSG_4).head(),
));
sproof.upsert_inbound_channel(ParaId::from(200)).mqc_head =
Some(MessageQueueChain::default().extend_hrmp(&MSG_4).head());
}
_ => unreachable!(),
},
@@ -1747,21 +1747,17 @@ mod tests {
#[test]
fn receive_hrmp_empty_channel() {
BlockTests::new()
.with_validation_data(
|_, relay_block_num, validation_data| match relay_block_num {
1 => {
// no channels
}
2 => {
// one new channel
validation_data.hrmp_mqc_heads.push((
ParaId::from(300),
MessageQueueChain::default().head(),
));
}
_ => unreachable!(),
},
)
.with_relay_sproof_builder(|_, relay_block_num, sproof| match relay_block_num {
1 => {
// no channels
}
2 => {
// one new channel
sproof.upsert_inbound_channel(ParaId::from(300)).mqc_head =
Some(MessageQueueChain::default().head());
}
_ => unreachable!(),
})
.add(1, || {})
.add(2, || {});
}
@@ -1783,30 +1779,24 @@ mod tests {
const ALICE: ParaId = ParaId::new(300);
BlockTests::new()
.with_validation_data(
|_, relay_block_num, validation_data| match relay_block_num {
.with_relay_sproof_builder(
|_, relay_block_num, sproof| match relay_block_num {
1 => {
validation_data.hrmp_mqc_heads.push((
ALICE,
MessageQueueChain::default().extend_hrmp(&MSG_1).head(),
));
sproof.upsert_inbound_channel(ALICE).mqc_head
= Some(MessageQueueChain::default().extend_hrmp(&MSG_1).head());
}
2 => {
// 300 - no new messages, mqc stayed the same.
validation_data.hrmp_mqc_heads.push((
ALICE,
MessageQueueChain::default().extend_hrmp(&MSG_1).head(),
));
sproof.upsert_inbound_channel(ALICE).mqc_head
= Some(MessageQueueChain::default().extend_hrmp(&MSG_1).head());
}
3 => {
// 300 - new message.
validation_data.hrmp_mqc_heads.push((
ALICE,
MessageQueueChain::default()
.extend_hrmp(&MSG_1)
.extend_hrmp(&MSG_2)
.head(),
));
sproof.upsert_inbound_channel(ALICE).mqc_head
= Some(MessageQueueChain::default()
.extend_hrmp(&MSG_1)
.extend_hrmp(&MSG_2)
.head());
}
_ => unreachable!(),
},
parachain-system/src/relay_state_snapshot.rs
+57 -9
View File
@@ -27,14 +27,27 @@ use sp_std::vec::Vec;
/// This data is essential for making sure that the parachain is aware of current resource use on
/// the relay chain and that the candidates produced for this parachain do not exceed any of these
/// limits.
#[derive(Encode, Decode)]
#[derive(Clone, Encode, Decode)]
pub struct MessagingStateSnapshot {
/// The current message queue chain head for downward message queue.
///
/// If the value is absent on the relay chain this will be set to all zeros.
pub dmq_mqc_head: relay_chain::Hash,
/// The current capacity of the upward message queue of the current parachain on the relay chain.
///
/// The capacity is represented by a tuple that consist of the `count` of the messages and the
/// `total_size` expressed as the sum of byte sizes of all messages in the queue.
pub relay_dispatch_queue_size: (u32, u32),
/// Information about all the inbound HRMP channels.
///
/// These are structured as a list of tuples. The para id in the tuple specifies the sender
/// of the channel. Obviously, the recipient is the current parachain.
///
/// The channels are sorted by the sender para id ascension.
pub ingress_channels: Vec<(ParaId, AbridgedHrmpChannel)>,
/// Information about all the outbound HRMP channels.
///
/// These are structured as a list of tuples. The para id in the tuple specifies the recipient
@@ -50,12 +63,16 @@ pub enum Error {
RootMismatch,
/// The host configuration cannot be extracted.
Config(ReadEntryErr),
/// The DMQ MQC head cannot be extracted.
DmqMqcHead(ReadEntryErr),
/// Relay dispatch queue cannot be extracted.
RelayDispatchQueueSize(ReadEntryErr),
/// The hrmp inress channel index cannot be extracted.
HrmpIngressChannelIndex(ReadEntryErr),
/// The hrmp egress channel index cannot be extracted.
HrmpEgressChannelIndex(ReadEntryErr),
/// The hrmp channel for the given recipient cannot be extracted.
HrmpChannel(ParaId, ReadEntryErr),
/// The channel identified by the sender and receiver cannot be extracted.
HrmpChannel(ParaId, ParaId, ReadEntryErr),
}
#[derive(Debug)]
@@ -92,14 +109,14 @@ where
/// of the current parachain and the expected storage root the proof should stem from.
pub fn extract_from_proof(
para_id: ParaId,
relay_storage_root: relay_chain::v1::Hash,
relay_parent_storage_root: relay_chain::v1::Hash,
proof: StorageProof,
) -> Result<(AbridgedHostConfiguration, MessagingStateSnapshot), Error> {
let db = proof.into_memory_db::<HashFor<relay_chain::Block>>();
if !db.contains(&relay_storage_root, EMPTY_PREFIX) {
if !db.contains(&relay_parent_storage_root, EMPTY_PREFIX) {
return Err(Error::RootMismatch);
}
let backend = TrieBackend::new(db, relay_storage_root);
let backend = TrieBackend::new(db, relay_parent_storage_root);
let host_config: AbridgedHostConfiguration = read_entry(
&backend,
@@ -108,6 +125,13 @@ pub fn extract_from_proof(
)
.map_err(Error::Config)?;
let dmq_mqc_head: relay_chain::Hash = read_entry(
&backend,
&relay_chain::well_known_keys::dmq_mqc_head(para_id),
Some(Default::default()),
)
.map_err(Error::DmqMqcHead)?;
let relay_dispatch_queue_size: (u32, u32) = read_entry(
&backend,
&relay_chain::well_known_keys::relay_dispatch_queue_size(para_id),
@@ -115,6 +139,13 @@ pub fn extract_from_proof(
)
.map_err(Error::RelayDispatchQueueSize)?;
let ingress_channel_index: Vec<ParaId> = read_entry(
&backend,
&relay_chain::well_known_keys::hrmp_ingress_channel_index(para_id),
Some(Vec::new()),
)
.map_err(Error::HrmpIngressChannelIndex)?;
let egress_channel_index: Vec<ParaId> = read_entry(
&backend,
&relay_chain::well_known_keys::hrmp_egress_channel_index(para_id),
@@ -122,6 +153,21 @@ pub fn extract_from_proof(
)
.map_err(Error::HrmpEgressChannelIndex)?;
let mut ingress_channels = Vec::with_capacity(ingress_channel_index.len());
for sender in ingress_channel_index {
let channel_id = relay_chain::v1::HrmpChannelId {
sender,
recipient: para_id,
};
let hrmp_channel: AbridgedHrmpChannel = read_entry(
&backend,
&relay_chain::well_known_keys::hrmp_channels(channel_id),
None,
)
.map_err(|read_err| Error::HrmpChannel(sender, para_id, read_err))?;
ingress_channels.push((sender, hrmp_channel));
}
let mut egress_channels = Vec::with_capacity(egress_channel_index.len());
for recipient in egress_channel_index {
let channel_id = relay_chain::v1::HrmpChannelId {
@@ -133,17 +179,19 @@ pub fn extract_from_proof(
&relay_chain::well_known_keys::hrmp_channels(channel_id),
None,
)
.map_err(|read_err| Error::HrmpChannel(recipient, read_err))?;
.map_err(|read_err| Error::HrmpChannel(para_id, recipient, read_err))?;
egress_channels.push((recipient, hrmp_channel));
}
// NOTE that egress_channels promises to be sorted. We satisfy this property by relying on
// the fact that `egress_channel_index` is itself sorted.
// NOTE that ingress_channels and egress_channels promise to be sorted. We satisfy this property
// by relying on the fact that `ingress_channel_index` and `egress_channel_index` are themselves sorted.
Ok((
host_config,
MessagingStateSnapshot {
dmq_mqc_head,
relay_dispatch_queue_size,
ingress_channels,
egress_channels,
},
))