mirror of
https://github.com/pezkuwichain/pezkuwi-subxt.git
synced 2026-04-26 19:17:58 +00:00
gossip: move authorities request to runtime api subsystem (#2798)
This commit is contained in:
@@ -19,9 +19,6 @@
|
||||
//! the gossiping subsystems on every new session.
|
||||
|
||||
use futures::{channel::mpsc, FutureExt as _};
|
||||
use std::sync::Arc;
|
||||
use sp_api::ProvideRuntimeApi;
|
||||
use sp_authority_discovery::AuthorityDiscoveryApi;
|
||||
use polkadot_node_subsystem::{
|
||||
messages::{
|
||||
GossipSupportMessage,
|
||||
@@ -34,7 +31,7 @@ use polkadot_node_subsystem_util::{
|
||||
self as util,
|
||||
};
|
||||
use polkadot_primitives::v1::{
|
||||
Hash, SessionIndex, AuthorityDiscoveryId, Block, BlockId,
|
||||
Hash, SessionIndex, AuthorityDiscoveryId,
|
||||
};
|
||||
use polkadot_node_network_protocol::{peer_set::PeerSet, PeerId};
|
||||
use sp_keystore::{CryptoStore, SyncCryptoStorePtr};
|
||||
@@ -43,8 +40,7 @@ use sp_application_crypto::{Public, AppKey};
|
||||
const LOG_TARGET: &str = "parachain::gossip-support";
|
||||
|
||||
/// The Gossip Support subsystem.
|
||||
pub struct GossipSupport<Client> {
|
||||
client: Arc<Client>,
|
||||
pub struct GossipSupport {
|
||||
keystore: SyncCryptoStorePtr,
|
||||
}
|
||||
|
||||
@@ -55,15 +51,10 @@ struct State {
|
||||
_last_connection_request: Option<mpsc::Receiver<(AuthorityDiscoveryId, PeerId)>>,
|
||||
}
|
||||
|
||||
impl<Client> GossipSupport<Client>
|
||||
where
|
||||
Client: ProvideRuntimeApi<Block>,
|
||||
Client::Api: AuthorityDiscoveryApi<Block>,
|
||||
{
|
||||
impl GossipSupport {
|
||||
/// Create a new instance of the [`GossipSupport`] subsystem.
|
||||
pub fn new(keystore: SyncCryptoStorePtr, client: Arc<Client>) -> Self {
|
||||
pub fn new(keystore: SyncCryptoStorePtr) -> Self {
|
||||
Self {
|
||||
client,
|
||||
keystore,
|
||||
}
|
||||
}
|
||||
@@ -74,7 +65,7 @@ where
|
||||
Context: SubsystemContext<Message = GossipSupportMessage>,
|
||||
{
|
||||
let mut state = State::default();
|
||||
let Self { client, keystore } = self;
|
||||
let Self { keystore } = self;
|
||||
loop {
|
||||
let message = match ctx.recv().await {
|
||||
Ok(message) => message,
|
||||
@@ -96,7 +87,7 @@ where
|
||||
tracing::trace!(target: LOG_TARGET, "active leaves signal");
|
||||
|
||||
let leaves = activated.into_iter().map(|a| a.hash);
|
||||
if let Err(e) = state.handle_active_leaves(&mut ctx, client.clone(), &keystore, leaves).await {
|
||||
if let Err(e) = state.handle_active_leaves(&mut ctx, &keystore, leaves).await {
|
||||
tracing::debug!(target: LOG_TARGET, error = ?e);
|
||||
}
|
||||
}
|
||||
@@ -109,18 +100,12 @@ where
|
||||
}
|
||||
}
|
||||
|
||||
async fn determine_relevant_authorities<Client>(
|
||||
client: Arc<Client>,
|
||||
async fn determine_relevant_authorities(
|
||||
ctx: &mut impl SubsystemContext,
|
||||
relay_parent: Hash,
|
||||
) -> Result<Vec<AuthorityDiscoveryId>, util::Error>
|
||||
where
|
||||
Client: ProvideRuntimeApi<Block>,
|
||||
Client::Api: AuthorityDiscoveryApi<Block>,
|
||||
{
|
||||
let api = client.runtime_api();
|
||||
let result = api.authorities(&BlockId::Hash(relay_parent))
|
||||
.map_err(|e| util::Error::RuntimeApi(format!("{:?}", e).into()));
|
||||
result
|
||||
) -> Result<Vec<AuthorityDiscoveryId>, util::Error> {
|
||||
let authorities = util::request_authorities_ctx(relay_parent, ctx).await?.await??;
|
||||
Ok(authorities)
|
||||
}
|
||||
|
||||
/// Return an error if we're not a validator in the given set (do not have keys).
|
||||
@@ -143,17 +128,12 @@ impl State {
|
||||
/// 1. Determine if the current session index has changed.
|
||||
/// 2. If it has, determine relevant validators
|
||||
/// and issue a connection request.
|
||||
async fn handle_active_leaves<Client>(
|
||||
async fn handle_active_leaves(
|
||||
&mut self,
|
||||
ctx: &mut impl SubsystemContext,
|
||||
client: Arc<Client>,
|
||||
keystore: &SyncCryptoStorePtr,
|
||||
leaves: impl Iterator<Item = Hash>,
|
||||
) -> Result<(), util::Error>
|
||||
where
|
||||
Client: ProvideRuntimeApi<Block>,
|
||||
Client::Api: AuthorityDiscoveryApi<Block>,
|
||||
{
|
||||
) -> Result<(), util::Error> {
|
||||
for leaf in leaves {
|
||||
let current_index = util::request_session_index_for_child_ctx(leaf, ctx).await?.await??;
|
||||
let maybe_new_session = match self.last_session_index {
|
||||
@@ -163,7 +143,7 @@ impl State {
|
||||
|
||||
if let Some((new_session, relay_parent)) = maybe_new_session {
|
||||
tracing::debug!(target: LOG_TARGET, %new_session, "New session detected");
|
||||
let authorities = determine_relevant_authorities(client.clone(), relay_parent).await?;
|
||||
let authorities = determine_relevant_authorities(ctx, relay_parent).await?;
|
||||
ensure_i_am_an_authority(keystore, &authorities).await?;
|
||||
tracing::debug!(target: LOG_TARGET, num = ?authorities.len(), "Issuing a connection request");
|
||||
|
||||
@@ -182,11 +162,9 @@ impl State {
|
||||
}
|
||||
}
|
||||
|
||||
impl<Client, Context> Subsystem<Context> for GossipSupport<Client>
|
||||
impl<Context> Subsystem<Context> for GossipSupport
|
||||
where
|
||||
Context: SubsystemContext<Message = GossipSupportMessage> + Sync + Send,
|
||||
Client: ProvideRuntimeApi<Block> + Send + 'static + Sync,
|
||||
Client::Api: AuthorityDiscoveryApi<Block>,
|
||||
{
|
||||
fn start(self, ctx: Context) -> SpawnedSubsystem {
|
||||
let future = self.run(ctx)
|
||||
|
||||
Reference in New Issue
Block a user