,
}
impl BackingGroupConnectionHelper {
pub fn new(keystore: pezsp_keystore::KeystorePtr, overseer_handle: OverseerHandle) -> Self {
Self { keystore, overseer_handle, our_slot: None }
}
async fn send_subsystem_message(&mut self, message: CollatorProtocolMessage) {
self.overseer_handle.send_msg(message, "BackingGroupConnectionHelper").await;
}
/// Update the current slot and initiate connections to backing groups if needed.
pub async fn update(&mut self, current_slot: Slot, authorities: &[P::Public])
where
P: pezsp_core::Pair + Send + Sync,
P::Public: Codec,
{
if Some(current_slot) <= self.our_slot {
// Current slot or next slot is ours.
// We already sent pre-connect message, no need to proceed further.
return;
}
let next_slot = current_slot + 1;
let next_slot_is_ours =
aura_internal::claim_slot::
(next_slot, authorities, &self.keystore)
.await
.is_some();
if next_slot_is_ours {
// Only send message if we were not connected. This avoids sending duplicate messages
// when running with a single collator.
if self.our_slot.is_none() {
// Next slot is ours, send connect message.
tracing::debug!(target: crate::LOG_TARGET, "Our slot {} is next, connecting to backing groups", next_slot);
self.send_subsystem_message(CollatorProtocolMessage::ConnectToBackingGroups)
.await;
}
self.our_slot = Some(next_slot);
} else if self.our_slot.take().is_some() {
// Next slot is not ours, send disconnect only if we had a slot before.
tracing::debug!(target: crate::LOG_TARGET, "Current slot = {}, disconnecting from backing groups", current_slot);
self.send_subsystem_message(CollatorProtocolMessage::DisconnectFromBackingGroups)
.await;
}
}
}
/// Check the `local_validation_code_hash` against the validation code hash in the relay chain
/// state.
///
/// If the code hashes do not match, it prints a warning.
async fn check_validation_code_or_log(
local_validation_code_hash: &ValidationCodeHash,
para_id: ParaId,
relay_client: &impl RelayChainInterface,
relay_parent: RelayHash,
) {
let state_validation_code_hash = match relay_client
.validation_code_hash(relay_parent, para_id, OccupiedCoreAssumption::Included)
.await
{
Ok(hash) => hash,
Err(error) => {
tracing::debug!(
target: super::LOG_TARGET,
%error,
?relay_parent,
%para_id,
"Failed to fetch validation code hash",
);
return;
},
};
match state_validation_code_hash {
Some(state) => {
if state != *local_validation_code_hash {
tracing::warn!(
target: super::LOG_TARGET,
%para_id,
?relay_parent,
?local_validation_code_hash,
relay_validation_code_hash = ?state,
"Teyrchain code doesn't match validation code stored in the relay chain state.",
);
}
},
None => {
tracing::warn!(
target: super::LOG_TARGET,
%para_id,
?relay_parent,
"Could not find validation code for teyrchain in the relay chain state.",
);
},
}
}
/// Fetch scheduling lookahead at given relay parent.
async fn scheduling_lookahead(
relay_parent: RelayHash,
relay_client: &impl RelayChainInterface,
) -> Option {
let runtime_api_version = relay_client
.version(relay_parent)
.await
.map_err(|e| {
tracing::error!(
target: super::LOG_TARGET,
error = ?e,
"Failed to fetch relay chain runtime version.",
)
})
.ok()?;
let teyrchain_host_runtime_api_version = runtime_api_version
.api_version(
&>::ID,
)
.unwrap_or_default();
if teyrchain_host_runtime_api_version
< RuntimeApiRequest::SCHEDULING_LOOKAHEAD_RUNTIME_REQUIREMENT
{
return None;
}
match relay_client.scheduling_lookahead(relay_parent).await {
Ok(scheduling_lookahead) => Some(scheduling_lookahead),
Err(err) => {
tracing::error!(
target: crate::LOG_TARGET,
?err,
?relay_parent,
"Failed to fetch scheduling lookahead from relay chain",
);
None
},
}
}
// Returns the claim queue at the given relay parent.
async fn claim_queue_at(
relay_parent: RelayHash,
relay_client: &impl RelayChainInterface,
) -> ClaimQueueSnapshot {
// Get `ClaimQueue` from runtime
match relay_client.claim_queue(relay_parent).await {
Ok(claim_queue) => claim_queue.into(),
Err(error) => {
tracing::error!(
target: crate::LOG_TARGET,
?error,
?relay_parent,
"Failed to query claim queue runtime API",
);
Default::default()
},
}
}
// Checks if we own the slot at the given block and whether there
// is space in the unincluded segment.
async fn can_build_upon(
para_slot: Slot,
relay_slot: Slot,
timestamp: Timestamp,
parent_hash: Block::Hash,
included_block: Block::Hash,
client: &Client,
keystore: &KeystorePtr,
) -> Option>
where
Client: ProvideRuntimeApi,
Client::Api: AuraApi + AuraUnincludedSegmentApi + ApiExt,
P: Pair,
P::Public: Codec,
P::Signature: Codec,
{
let runtime_api = client.runtime_api();
let authorities = runtime_api.authorities(parent_hash).ok()?;
let author_pub = aura_internal::claim_slot::(para_slot, &authorities, keystore).await?;
// This function is typically called when we want to build block N. At that point, the
// unincluded segment in the runtime is unaware of the hash of block N-1. If the unincluded
// segment in the runtime is full, but block N-1 is the included block, the unincluded segment
// should have length 0 and we can build. Since the hash is not available to the runtime
// however, we need this extra check here.
if parent_hash == included_block {
return Some(SlotClaim::unchecked::
(author_pub, para_slot, timestamp));
}
let api_version = runtime_api
.api_version::>(parent_hash)
.ok()
.flatten()?;
let slot = if api_version > 1 { relay_slot } else { para_slot };
runtime_api
.can_build_upon(parent_hash, included_block, slot)
.ok()?
.then(|| SlotClaim::unchecked::(author_pub, para_slot, timestamp))
}
/// Use [`pezcumulus_client_consensus_common::find_potential_parents`] to find teyrchain blocks that
/// we can build on. Once a list of potential parents is retrieved, return the last one of the
/// longest chain.
async fn find_parent(
relay_parent: RelayHash,
para_id: ParaId,
para_backend: &impl pezsc_client_api::Backend,
relay_client: &impl RelayChainInterface,
) -> Option<(::Header, consensus_common::PotentialParent)>
where
Block: BlockT,
{
let parent_search_params = ParentSearchParams {
relay_parent,
para_id,
ancestry_lookback: scheduling_lookahead(relay_parent, relay_client)
.await
.unwrap_or(DEFAULT_SCHEDULING_LOOKAHEAD)
.saturating_sub(1) as usize,
max_depth: PARENT_SEARCH_DEPTH,
ignore_alternative_branches: true,
};
let potential_parents = pezcumulus_client_consensus_common::find_potential_parents::(
parent_search_params,
para_backend,
relay_client,
)
.await;
let potential_parents = match potential_parents {
Err(e) => {
tracing::error!(
target: crate::LOG_TARGET,
?relay_parent,
err = ?e,
"Could not fetch potential parents to build upon"
);
return None;
},
Ok(x) => x,
};
let included_block = potential_parents.iter().find(|x| x.depth == 0)?.header.clone();
potential_parents
.into_iter()
.max_by_key(|a| a.depth)
.map(|parent| (included_block, parent))
}
#[cfg(test)]
mod tests {
use super::*;
use crate::collators::{can_build_upon, BackingGroupConnectionHelper};
use codec::Encode;
use futures::StreamExt;
use pezcumulus_primitives_aura::Slot;
use pezcumulus_primitives_core::BlockT;
use pezcumulus_relay_chain_interface::PHash;
use pezcumulus_test_client::{
runtime::{Block, Hash},
Client, DefaultTestClientBuilderExt, InitBlockBuilder, TestClientBuilder,
TestClientBuilderExt,
};
use pezcumulus_test_relay_sproof_builder::RelayStateSproofBuilder;
use pezkuwi_overseer::{Event, Handle};
use pezkuwi_primitives::HeadData;
use pezsc_consensus::{BlockImport, BlockImportParams, ForkChoiceStrategy};
use pezsp_consensus::BlockOrigin;
use pezsp_keystore::{Keystore, KeystorePtr};
use pezsp_timestamp::Timestamp;
use std::sync::{Arc, Mutex};
async fn import_block>(
importer: &I,
block: Block,
origin: BlockOrigin,
import_as_best: bool,
) {
let (header, body) = block.deconstruct();
let mut block_import_params = BlockImportParams::new(origin, header);
block_import_params.fork_choice = Some(ForkChoiceStrategy::Custom(import_as_best));
block_import_params.body = Some(body);
importer.import_block(block_import_params).await.unwrap();
}
fn sproof_with_parent_by_hash(client: &Client, hash: PHash) -> RelayStateSproofBuilder {
let header = client.header(hash).ok().flatten().expect("No header for parent block");
let included = HeadData(header.encode());
let mut builder = RelayStateSproofBuilder::default();
builder.para_id = pezcumulus_test_client::runtime::TEYRCHAIN_ID.into();
builder.included_para_head = Some(included);
builder
}
async fn build_and_import_block(client: &Client, included: Hash) -> Block {
let sproof = sproof_with_parent_by_hash(client, included);
let block_builder = client.init_block_builder(None, sproof).block_builder;
let block = block_builder.build().unwrap().block;
let origin = BlockOrigin::NetworkInitialSync;
import_block(client, block.clone(), origin, true).await;
block
}
fn set_up_components(num_authorities: usize) -> (Arc, KeystorePtr) {
let keystore = Arc::new(pezsp_keystore::testing::MemoryKeystore::new()) as Arc<_>;
for key in pezsp_keyring::Sr25519Keyring::iter().take(num_authorities) {
Keystore::sr25519_generate_new(
&*keystore,
pezsp_application_crypto::key_types::AURA,
Some(&key.to_seed()),
)
.expect("Can insert key into MemoryKeyStore");
}
(Arc::new(TestClientBuilder::new().build()), keystore)
}
/// This tests a special scenario where the unincluded segment in the runtime
/// is full. We are calling `can_build_upon`, passing the last built block as the
/// included one. In the runtime we will not find the hash of the included block in the
/// unincluded segment. The `can_build_upon` runtime API would therefore return `false`, but
/// we are ensuring on the node side that we are are always able to build on the included block.
#[tokio::test]
async fn test_can_build_upon() {
let (client, keystore) = set_up_components(6);
let genesis_hash = client.chain_info().genesis_hash;
let mut last_hash = genesis_hash;
// Fill up the unincluded segment tracker in the runtime.
while can_build_upon::<_, _, pezsp_consensus_aura::sr25519::AuthorityPair>(
Slot::from(u64::MAX),
Slot::from(u64::MAX),
Timestamp::default(),
last_hash,
genesis_hash,
&*client,
&keystore,
)
.await
.is_some()
{
let block = build_and_import_block(&client, genesis_hash).await;
last_hash = block.header().hash();
}
// Blocks were built with the genesis hash set as included block.
// We call `can_build_upon` with the last built block as the included block.
let result = can_build_upon::<_, _, pezsp_consensus_aura::sr25519::AuthorityPair>(
Slot::from(u64::MAX),
Slot::from(u64::MAX),
Timestamp::default(),
last_hash,
last_hash,
&*client,
&keystore,
)
.await;
assert!(result.is_some());
}
/// Helper to create a mock overseer handle and message recorder
fn create_overseer_handle() -> (OverseerHandle, Arc>>) {
let messages = Arc::new(Mutex::new(Vec::new()));
let messages_clone = messages.clone();
let (tx, mut rx) = pezkuwi_node_subsystem_util::metered::channel(100);
// Spawn a task to receive and record overseer messages
tokio::spawn(async move {
while let Some(event) = rx.next().await {
if let Event::MsgToSubsystem { msg, .. } = event {
if let pezkuwi_node_subsystem::AllMessages::CollatorProtocol(cp_msg) = msg {
messages_clone.lock().unwrap().push(cp_msg);
}
}
}
});
(Handle::new(tx), messages)
}
#[tokio::test]
async fn preconnect_when_next_slot_is_ours() {
let (client, keystore) = set_up_components(1);
let genesis_hash = client.chain_info().genesis_hash;
let (overseer_handle, messages_recorder) = create_overseer_handle();
let mut helper = BackingGroupConnectionHelper::new(keystore, overseer_handle);
// Fetch authorities for the update call
let authorities = client.runtime_api().authorities(genesis_hash).unwrap();
// Update with slot 5, next slot (6) should be ours
helper
.update::(Slot::from(5), &authorities)
.await;
// Give time for message to be processed
tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
let messages = messages_recorder.lock().unwrap();
assert_eq!(messages.len(), 1);
assert!(matches!(messages[0], CollatorProtocolMessage::ConnectToBackingGroups));
assert_eq!(helper.our_slot, Some(Slot::from(6)));
}
#[tokio::test]
async fn preconnect_no_duplicate_connect_message() {
let (client, keystore) = set_up_components(1);
let genesis_hash = client.chain_info().genesis_hash;
let (overseer_handle, messages_recorder) = create_overseer_handle();
let mut helper = BackingGroupConnectionHelper::new(keystore, overseer_handle);
// Fetch authorities for the update calls
let authorities = client.runtime_api().authorities(genesis_hash).unwrap();
// Update with slot 5, next slot (6) is ours
helper
.update::(Slot::from(5), &authorities)
.await;
// Give time for message to be processed
tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
assert_eq!(messages_recorder.lock().unwrap().len(), 1);
messages_recorder.lock().unwrap().clear();
// Update with slot 5 again - should not send another message
helper
.update::(Slot::from(5), &authorities)
.await;
tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
assert_eq!(messages_recorder.lock().unwrap().len(), 0);
// Update with slot 1 (our slot) - should not send another message
helper
.update::(Slot::from(6), &authorities)
.await;
tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
assert_eq!(messages_recorder.lock().unwrap().len(), 0);
}
#[tokio::test]
async fn preconnect_disconnect_when_slot_passes() {
let (client, keystore) = set_up_components(1);
let genesis_hash = client.chain_info().genesis_hash;
let (overseer_handle, messages_recorder) = create_overseer_handle();
let mut helper = BackingGroupConnectionHelper::new(keystore, overseer_handle);
// Fetch authorities for the update calls
let authorities = client.runtime_api().authorities(genesis_hash).unwrap();
// Slot 0 -> Alice, Slot 1 -> Bob, Slot 2 -> Charlie, Slot 3 -> Dave, Slot 4 -> Eve,
// Slot 5 -> Ferdie, Slot 6 -> Alice
// Update with slot 5, next slot (6) is ours -> should connect
helper
.update::(Slot::from(5), &authorities)
.await;
tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
assert_eq!(helper.our_slot, Some(Slot::from(6)));
messages_recorder.lock().unwrap().clear();
// Update with slot 8, next slot (9) is Charlie's -> should disconnect
helper
.update::(Slot::from(8), &authorities)
.await;
tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
{
let messages = messages_recorder.lock().unwrap();
assert_eq!(messages.len(), 1, "Expected exactly one disconnect message");
assert!(matches!(messages[0], CollatorProtocolMessage::DisconnectFromBackingGroups));
assert_eq!(helper.our_slot, None);
}
messages_recorder.lock().unwrap().clear();
// Update again with slot 8, next slot (9) is Charlie's -> should not send another
// disconnect message
helper
.update::(Slot::from(8), &authorities)
.await;
tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
let messages = messages_recorder.lock().unwrap();
assert_eq!(messages.len(), 0, "Expected no messages");
assert_eq!(helper.our_slot, None);
}
#[tokio::test]
async fn preconnect_no_disconnect_without_previous_connection() {
let (client, keystore) = set_up_components(1);
let genesis_hash = client.chain_info().genesis_hash;
let (overseer_handle, messages_recorder) = create_overseer_handle();
let mut helper = BackingGroupConnectionHelper::new(keystore, overseer_handle);
// Fetch authorities for the update call
let authorities = client.runtime_api().authorities(genesis_hash).unwrap();
// Slot 0 -> Alice, Slot 1 -> Bob, Slot 2 -> Charlie, Slot 3 -> Dave, Slot 4 -> Eve,
// Slot 5 -> Ferdie
// Update with slot 1 (Bob's slot), next slot (2) is Charlie's
// Since we never connected before (our_slot is None), we should not send disconnect
helper
.update::(Slot::from(1), &authorities)
.await;
tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
// Should not send any message since we never connected
assert_eq!(messages_recorder.lock().unwrap().len(), 0);
assert_eq!(helper.our_slot, None);
}
#[tokio::test]
async fn preconnect_multiple_cycles() {
let (client, keystore) = set_up_components(1);
let genesis_hash = client.chain_info().genesis_hash;
let (overseer_handle, messages_recorder) = create_overseer_handle();
let mut helper = BackingGroupConnectionHelper::new(keystore, overseer_handle);
// Fetch authorities for the update calls
let authorities = client.runtime_api().authorities(genesis_hash).unwrap();
// Slot 0 -> Alice, Slot 1 -> Bob, Slot 2 -> Charlie, Slot 3 -> Dave, Slot 4 -> Eve,
// Slot 5 -> Ferdie, Slot 6 -> Alice, Slot 7 -> Bob, ...
// Cycle 1: Connect at slot 5, next slot (6) is ours
helper
.update::(Slot::from(5), &authorities)
.await;
tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
{
let messages = messages_recorder.lock().unwrap();
assert_eq!(messages.len(), 1);
assert!(matches!(messages[0], CollatorProtocolMessage::ConnectToBackingGroups));
}
assert_eq!(helper.our_slot, Some(Slot::from(6)));
messages_recorder.lock().unwrap().clear();
// Cycle 1: Disconnect at slot 7, next slot (8) is Charlie's
helper
.update::(Slot::from(7), &authorities)
.await;
tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
{
let messages = messages_recorder.lock().unwrap();
assert_eq!(messages.len(), 1);
assert!(matches!(messages[0], CollatorProtocolMessage::DisconnectFromBackingGroups));
}
assert_eq!(helper.our_slot, None);
messages_recorder.lock().unwrap().clear();
// Cycle 2: Connect again at slot 11, next slot (12) is ours
helper
.update::(Slot::from(11), &authorities)
.await;
tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
{
let messages = messages_recorder.lock().unwrap();
assert_eq!(messages.len(), 1);
assert!(matches!(messages[0], CollatorProtocolMessage::ConnectToBackingGroups));
}
assert_eq!(helper.our_slot, Some(Slot::from(12)));
}
#[tokio::test]
async fn preconnect_handles_empty_authorities() {
let keystore = Arc::new(pezsp_keystore::testing::MemoryKeystore::new()) as Arc<_>;
let (overseer_handle, messages_recorder) = create_overseer_handle();
let mut helper = BackingGroupConnectionHelper::new(keystore, overseer_handle);
// Pass empty authorities list
let authorities = vec![];
helper
.update::(Slot::from(0), &authorities)
.await;
tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
// Should not send any message if authorities list is empty
assert_eq!(messages_recorder.lock().unwrap().len(), 0);
}
}
/// Holds a relay parent and its descendants.
pub struct RelayParentData {
/// The relay parent block header
relay_parent: RelayHeader,
/// Ordered collection of descendant block headers, from oldest to newest
descendants: Vec,
}
impl RelayParentData {
/// Creates a new instance with the given relay parent and no descendants.
pub fn new(relay_parent: RelayHeader) -> Self {
Self { relay_parent, descendants: Default::default() }
}
/// Creates a new instance with the given relay parent and descendants.
pub fn new_with_descendants(relay_parent: RelayHeader, descendants: Vec) -> Self {
Self { relay_parent, descendants }
}
/// Returns a reference to the relay parent header.
pub fn relay_parent(&self) -> &RelayHeader {
&self.relay_parent
}
/// Returns the number of descendants.
#[cfg(test)]
pub fn descendants_len(&self) -> usize {
self.descendants.len()
}
/// Consumes the structure and returns a vector containing the relay parent followed by its
/// descendants in chronological order. The resulting list should be provided to the teyrchain
/// inherent data.
pub fn into_inherent_descendant_list(self) -> Vec {
let Self { relay_parent, mut descendants } = self;
if descendants.is_empty() {
return Default::default();
}
let mut result = vec![relay_parent];
result.append(&mut descendants);
result
}
}