From 103a36abd67612559caef1157ecefa0157a11ae2 Mon Sep 17 00:00:00 2001 From: Pierre Krieger Date: Mon, 12 Jan 2026 12:03:45 +0100 Subject: [PATCH 1/3] Rename standalone chain to Substrate-compatible chain --- light-base/src/lib.rs | 42 +++++++++---------- light-base/src/sync_service.rs | 28 ++++++------- .../{standalone.rs => substrate_compat.rs} | 11 ++--- 3 files changed, 40 insertions(+), 41 deletions(-) rename light-base/src/sync_service/{standalone.rs => substrate_compat.rs} (99%) diff --git a/light-base/src/lib.rs b/light-base/src/lib.rs index edab110fed..11a03b1c05 100644 --- a/light-base/src/lib.rs +++ b/light-base/src/lib.rs @@ -714,7 +714,7 @@ impl Client { } } (None, Some(chain_information)) => { - StartServicesChainTy::RelayChain { chain_information } + StartServicesChainTy::SubstrateCompatible { chain_information } } (None, None) => { // Checked above. @@ -1110,7 +1110,7 @@ pub enum AddChainError { } enum StartServicesChainTy<'a, TPlat: platform::PlatformRef> { - RelayChain { + SubstrateCompatible { chain_information: &'a chain::chain_information::ValidChainInformation, }, Parachain { @@ -1148,27 +1148,25 @@ fn start_services( let network_service_chain = network_service.add_chain(network_service::ConfigChain { log_name: log_name.clone(), num_out_slots: 4, - grandpa_protocol_finalized_block_height: if let StartServicesChainTy::RelayChain { - chain_information, - } = &config - { - if matches!( - chain_information.as_ref().finality, - chain::chain_information::ChainInformationFinalityRef::Grandpa { .. } - ) { - Some(chain_information.as_ref().finalized_block_header.number) + grandpa_protocol_finalized_block_height: + if let StartServicesChainTy::SubstrateCompatible { chain_information } = &config { + if matches!( + chain_information.as_ref().finality, + chain::chain_information::ChainInformationFinalityRef::Grandpa { .. } + ) { + Some(chain_information.as_ref().finalized_block_header.number) + } else { + None + } } else { + // Parachains never use GrandPa. None - } - } else { - // Parachains never use GrandPa. - None - }, + }, genesis_block_hash: header::hash_from_scale_encoded_header( &genesis_block_scale_encoded_header, ), best_block: match &config { - StartServicesChainTy::RelayChain { chain_information } => ( + StartServicesChainTy::SubstrateCompatible { chain_information } => ( chain_information.as_ref().finalized_block_header.number, chain_information .as_ref() @@ -1236,8 +1234,8 @@ fn start_services( (sync_service, runtime_service) } - StartServicesChainTy::RelayChain { chain_information } => { - // Chain is a relay chain. + StartServicesChainTy::SubstrateCompatible { chain_information } => { + // Chain is a Substrate-compatible non-parachain chain. // The sync service is leveraging the network service, downloads block headers, // and verifies them, to determine what are the best and finalized blocks of the @@ -1247,11 +1245,11 @@ fn start_services( block_number_bytes, platform: platform.clone(), network_service: network_service_chain.clone(), - chain_type: sync_service::ConfigChainType::RelayChain( - sync_service::ConfigRelayChain { + chain_type: sync_service::ConfigChainType::SubstrateCompatible( + sync_service::ConfigSubstrateCompatible { chain_information: chain_information.clone(), runtime_code_hint: runtime_code_hint.map(|hint| { - sync_service::ConfigRelayChainRuntimeCodeHint { + sync_service::ConfigSubstrateCompatibleRuntimeCodeHint { storage_value: hint.code, merkle_value: hint.code_merkle_value, closest_ancestor_excluding: hint.closest_ancestor_excluding, diff --git a/light-base/src/sync_service.rs b/light-base/src/sync_service.rs index c45e7d5306..6eb2f20df3 100644 --- a/light-base/src/sync_service.rs +++ b/light-base/src/sync_service.rs @@ -45,7 +45,7 @@ use smoldot::{ }; mod parachain; -mod standalone; +mod substrate_compat; pub use network_service::Role; @@ -66,20 +66,20 @@ pub struct Config { /// Access to the network, and index of the chain to sync from the point of view of the /// network service. pub network_service: Arc>, - /// Extra fields depending on whether the chain is a relay chain or a parachain. + /// Extra fields depending on whether the chain is a parachain. pub chain_type: ConfigChainType, } /// See [`Config::chain_type`]. pub enum ConfigChainType { - /// Chain is a relay chain. - RelayChain(ConfigRelayChain), + /// Chain is a Substrate-compatible non-parachain. + SubstrateCompatible(ConfigSubstrateCompatible), /// Chain is a parachain. Parachain(ConfigParachain), } -/// See [`ConfigChainType::RelayChain`]. -pub struct ConfigRelayChain { +/// See [`ConfigChainType::SubstrateCompatible`]. +pub struct ConfigSubstrateCompatible { /// State of the finalized chain. pub chain_information: chain::chain_information::ValidChainInformation, @@ -89,13 +89,13 @@ pub struct ConfigRelayChain { /// if it matches the Merkle value provided in the hint, use the storage value in the hint /// instead of downloading it. If the hint doesn't match, an extra round-trip will be needed, /// but if the hint matches it saves a big download. - pub runtime_code_hint: Option, + pub runtime_code_hint: Option, } -/// See [`ConfigRelayChain::runtime_code_hint`]. -pub struct ConfigRelayChainRuntimeCodeHint { +/// See [`ConfigSubstrateCompatible::runtime_code_hint`]. +pub struct ConfigSubstrateCompatibleRuntimeCodeHint { /// Storage value of the `:code` trie node corresponding to - /// [`ConfigRelayChainRuntimeCodeHint::merkle_value`]. + /// [`ConfigSubstrateCompatibleRuntimeCodeHint::merkle_value`]. pub storage_value: Vec, /// Merkle value of the `:code` trie node in the storage main trie. pub merkle_value: Vec, @@ -154,13 +154,13 @@ impl SyncService { from_foreground, config.network_service.clone(), )), - ConfigChainType::RelayChain(config_relay_chain) => { - Box::pin(standalone::start_standalone_chain( + ConfigChainType::SubstrateCompatible(config_substrate_compat) => { + Box::pin(substrate_compat::start_substrate_compatible_chain( log_target.clone(), config.platform.clone(), - config_relay_chain.chain_information, + config_substrate_compat.chain_information, config.block_number_bytes, - config_relay_chain.runtime_code_hint, + config_substrate_compat.runtime_code_hint, from_foreground, config.network_service.clone(), )) diff --git a/light-base/src/sync_service/standalone.rs b/light-base/src/sync_service/substrate_compat.rs similarity index 99% rename from light-base/src/sync_service/standalone.rs rename to light-base/src/sync_service/substrate_compat.rs index b5d39b75a0..1cf52b0590 100644 --- a/light-base/src/sync_service/standalone.rs +++ b/light-base/src/sync_service/substrate_compat.rs @@ -16,8 +16,8 @@ // along with this program. If not, see . use super::{ - BlockNotification, ConfigRelayChainRuntimeCodeHint, FinalizedBlockRuntime, Notification, - SubscribeAll, ToBackground, + BlockNotification, ConfigSubstrateCompatibleRuntimeCodeHint, FinalizedBlockRuntime, + Notification, SubscribeAll, ToBackground, }; use crate::{log, network_service, platform::PlatformRef, util}; @@ -41,13 +41,14 @@ use smoldot::{ sync::all, }; -/// Starts a sync service background task to synchronize a standalone chain (relay chain or not). -pub(super) async fn start_standalone_chain( +/// Starts a sync service background task to synchronize a chain (relay chain or not) that is +/// built with Substrate. +pub(super) async fn start_substrate_compatible_chain( log_target: String, platform: TPlat, chain_information: chain::chain_information::ValidChainInformation, block_number_bytes: usize, - runtime_code_hint: Option, + runtime_code_hint: Option, mut from_foreground: Pin>>, network_service: Arc>, ) { From c4bf89ce094282cf34ea15a39ddd046acad0ce18 Mon Sep 17 00:00:00 2001 From: Pierre Krieger Date: Mon, 12 Jan 2026 12:14:10 +0100 Subject: [PATCH 2/3] Update sync service API for smart parachains --- light-base/src/lib.rs | 7 ++++-- light-base/src/sync_service.rs | 25 ++++++++++++++----- .../src/sync_service/substrate_compat.rs | 3 ++- 3 files changed, 26 insertions(+), 9 deletions(-) diff --git a/light-base/src/lib.rs b/light-base/src/lib.rs index 11a03b1c05..5443ffa202 100644 --- a/light-base/src/lib.rs +++ b/light-base/src/lib.rs @@ -1214,8 +1214,10 @@ fn start_services( chain_type: sync_service::ConfigChainType::Parachain( sync_service::ConfigParachain { finalized_block_header, - para_id, - relay_chain_sync: relay_chain.runtime_service.clone(), + relay_chain: sync_service::ConfigRelayChain { + para_id, + relay_chain_sync: relay_chain.runtime_service.clone(), + }, }, ), })); @@ -1255,6 +1257,7 @@ fn start_services( closest_ancestor_excluding: hint.closest_ancestor_excluding, } }), + relay_chain: None, }, ), })); diff --git a/light-base/src/sync_service.rs b/light-base/src/sync_service.rs index 6eb2f20df3..1421bdacbf 100644 --- a/light-base/src/sync_service.rs +++ b/light-base/src/sync_service.rs @@ -73,13 +73,13 @@ pub struct Config { /// See [`Config::chain_type`]. pub enum ConfigChainType { /// Chain is a Substrate-compatible non-parachain. - SubstrateCompatible(ConfigSubstrateCompatible), + SubstrateCompatible(ConfigSubstrateCompatible), /// Chain is a parachain. Parachain(ConfigParachain), } /// See [`ConfigChainType::SubstrateCompatible`]. -pub struct ConfigSubstrateCompatible { +pub struct ConfigSubstrateCompatible { /// State of the finalized chain. pub chain_information: chain::chain_information::ValidChainInformation, @@ -90,6 +90,10 @@ pub struct ConfigSubstrateCompatible { /// instead of downloading it. If the hint doesn't match, an extra round-trip will be needed, /// but if the hint matches it saves a big download. pub runtime_code_hint: Option, + + /// If this chain is a parachain, contains the information of the relay chain. + /// `None` if this chain isn't a parachain. + pub relay_chain: Option>, } /// See [`ConfigSubstrateCompatible::runtime_code_hint`]. @@ -105,13 +109,19 @@ pub struct ConfigSubstrateCompatibleRuntimeCodeHint { /// See [`ConfigChainType::Parachain`]. pub struct ConfigParachain { - /// Runtime service that synchronizes the relay chain of this parachain. - pub relay_chain_sync: Arc>, + /// Parameters of the relay chain. + pub relay_chain: ConfigRelayChain, /// SCALE-encoded header of a known finalized block of the parachain. Used in the situation /// where the API user subscribes using [`SyncService::subscribe_all`] before any parachain /// block can be gathered. pub finalized_block_header: Vec, +} + +/// See [`ConfigSubstrateCompatible::relay_chain`] and [`ConfigParachain::relay_chain`]. +pub struct ConfigRelayChain { + /// Runtime service that synchronizes the relay chain of this parachain. + pub relay_chain_sync: Arc>, /// Id of the parachain within the relay chain. /// @@ -149,8 +159,8 @@ impl SyncService { config.platform.clone(), config_parachain.finalized_block_header, config.block_number_bytes, - config_parachain.relay_chain_sync.clone(), - config_parachain.para_id, + config_parachain.relay_chain.relay_chain_sync.clone(), + config_parachain.relay_chain.para_id, from_foreground, config.network_service.clone(), )), @@ -160,6 +170,9 @@ impl SyncService { config.platform.clone(), config_substrate_compat.chain_information, config.block_number_bytes, + config_substrate_compat + .relay_chain + .map(|rc| (rc.relay_chain_sync, rc.para_id)), config_substrate_compat.runtime_code_hint, from_foreground, config.network_service.clone(), diff --git a/light-base/src/sync_service/substrate_compat.rs b/light-base/src/sync_service/substrate_compat.rs index 1cf52b0590..320a661f97 100644 --- a/light-base/src/sync_service/substrate_compat.rs +++ b/light-base/src/sync_service/substrate_compat.rs @@ -19,7 +19,7 @@ use super::{ BlockNotification, ConfigSubstrateCompatibleRuntimeCodeHint, FinalizedBlockRuntime, Notification, SubscribeAll, ToBackground, }; -use crate::{log, network_service, platform::PlatformRef, util}; +use crate::{log, network_service, platform::PlatformRef, runtime_service, util}; use alloc::{ borrow::{Cow, ToOwned as _}, @@ -48,6 +48,7 @@ pub(super) async fn start_substrate_compatible_chain( platform: TPlat, chain_information: chain::chain_information::ValidChainInformation, block_number_bytes: usize, + _relay_chain: Option<(Arc>, u32)>, runtime_code_hint: Option, mut from_foreground: Pin>>, network_service: Arc>, From b526b4c9bf130130e7f03fd85346f0ba7babe4ef Mon Sep 17 00:00:00 2001 From: Pierre Krieger Date: Wed, 14 Jan 2026 14:36:24 +0100 Subject: [PATCH 3/3] Extract a paraheads service from the parachains syncing --- light-base/src/sync_service.rs | 1 + light-base/src/sync_service/parachain.rs | 1223 +++------------------ light-base/src/sync_service/paraheads.rs | 1235 ++++++++++++++++++++++ 3 files changed, 1407 insertions(+), 1052 deletions(-) create mode 100644 light-base/src/sync_service/paraheads.rs diff --git a/light-base/src/sync_service.rs b/light-base/src/sync_service.rs index 1421bdacbf..2fe1e59ceb 100644 --- a/light-base/src/sync_service.rs +++ b/light-base/src/sync_service.rs @@ -45,6 +45,7 @@ use smoldot::{ }; mod parachain; +mod paraheads; mod substrate_compat; pub use network_service::Role; diff --git a/light-base/src/sync_service/parachain.rs b/light-base/src/sync_service/parachain.rs index 38d3b898c4..bca9e61cd8 100644 --- a/light-base/src/sync_service/parachain.rs +++ b/light-base/src/sync_service/parachain.rs @@ -15,18 +15,16 @@ // You should have received a copy of the GNU General Public License // along with this program. If not, see . -use super::ToBackground; +use super::{ToBackground, paraheads}; use crate::{log, network_service, platform::PlatformRef, runtime_service, util}; -use alloc::{borrow::ToOwned as _, boxed::Box, format, string::String, sync::Arc, vec::Vec}; -use core::{mem, num::NonZero, pin::Pin, time::Duration}; +use alloc::{boxed::Box, format, string::String, sync::Arc, vec::Vec}; +use core::{iter, pin::Pin}; +use futures_channel::oneshot; use futures_lite::FutureExt as _; -use futures_util::{StreamExt as _, future, stream}; +use futures_util::{StreamExt as _, future}; use hashbrown::HashMap; -use itertools::Itertools as _; -use smoldot::{ - chain::async_tree, header, informant::HashDisplay, libp2p::PeerId, network::codec, sync::para, -}; +use smoldot::{header, informant::HashDisplay, libp2p::PeerId, network::codec}; /// Starts a sync service background task to synchronize a parachain. pub(super) async fn start_parachain( @@ -39,14 +37,34 @@ pub(super) async fn start_parachain( from_foreground: Pin>>, network_service: Arc>, ) { + let (to_paraheads, from_paraheads) = async_channel::bounded(16); + let from_paraheads = Box::pin(from_paraheads); + + let paraheads_log_target = format!("{log_target}-paraheads"); + platform.spawn_task(paraheads_log_target.clone().into(), { + let platform = platform.clone(); + let task = paraheads::start_paraheads( + paraheads_log_target.clone(), + platform.clone(), + finalized_block_header, + relay_chain_sync, + parachain_id, + from_paraheads, + ); + + async move { + task.await; + log!(&platform, Debug, ¶heads_log_target, "paraheads-hutdown"); + } + }); + ParachainBackgroundTask { log_target, from_foreground, block_number_bytes, - parachain_id, + paraheads: to_paraheads.clone(), from_network_service: None, network_service, - obsolete_finalized_parahead: finalized_block_header, sync_sources: HashMap::with_capacity_and_hasher( 0, util::SipHasherBuild::new({ @@ -56,17 +74,21 @@ pub(super) async fn start_parachain( }), ), subscription_state: ParachainBackgroundState::NotSubscribed { - all_subscriptions: Vec::new(), subscribe_future: { - let relay_chain_sync = relay_chain_sync.clone(); Box::pin(async move { - relay_chain_sync - .subscribe_all(32, NonZero::::new(usize::MAX).unwrap()) - .await + let (send_back, sub_rx) = oneshot::channel(); + let _ = to_paraheads + .send(super::ToBackground::SubscribeAll { + send_back, + buffer_size: 32, + runtime_interest: false, + }) + .await; + // TODO: don't panic if the paraheads service dies here + sub_rx.await.unwrap() }) }, }, - relay_chain_sync, platform, } .run() @@ -87,8 +109,8 @@ struct ParachainBackgroundTask { /// Number of bytes to use to encode the parachain block numbers in headers. block_number_bytes: usize, - /// Id of the parachain registered within the relay chain. Chosen by the user. - parachain_id: u32, + /// Channel to the paraheads background service. + paraheads: async_channel::Sender, /// Networking service connected to the peer-to-peer network of the parachain. network_service: Arc>, @@ -96,15 +118,6 @@ struct ParachainBackgroundTask { /// Events coming from the networking service. `None` if not subscribed yet. from_network_service: Option>>>, - /// Runtime service of the relay chain. - relay_chain_sync: Arc>, - - /// Last-known finalized parachain header. Can be very old and obsolete. - /// Updated after we successfully fetch the parachain head of a relay chain finalized block, - /// and left untouched if the fetch fails. - /// Initialized to the parachain genesis block header. - obsolete_finalized_parahead: Vec, - /// List of parachain network sources. /// /// Values are their role, and self-reported best block when we connected to them. This best @@ -117,80 +130,34 @@ struct ParachainBackgroundTask { /// > all parachain nodes know about all parachain blocks from the relay chain. sync_sources: HashMap, - /// Extra fields that are set after the subscription to the runtime service events has + /// Extra fields that are set after the subscription to the paraheads service events has /// succeeded. - subscription_state: ParachainBackgroundState, + subscription_state: ParachainBackgroundState, } -enum ParachainBackgroundState { - /// Currently subscribing to the relay chain runtime service. +enum ParachainBackgroundState { + /// Currently subscribing to the paraheads service. NotSubscribed { - /// List of senders that will get notified when the tree of blocks is modified. - /// - /// These subscriptions are pending and no notification should be sent to them until the - /// subscription to the relay chain runtime service is finished. - all_subscriptions: Vec>, - /// Future when the subscription has finished. - subscribe_future: future::BoxFuture<'static, runtime_service::SubscribeAll>, + subscribe_future: future::BoxFuture<'static, super::SubscribeAll>, }, - /// Subscribed to the relay chain runtime service. - Subscribed(ParachainBackgroundTaskAfterSubscription), + /// Subscribed to the paraheads service. + Subscribed(ParachainBackgroundTaskAfterSubscription), } -struct ParachainBackgroundTaskAfterSubscription { - /// List of senders that get notified when the tree of blocks is modified. - all_subscriptions: Vec>, - +struct ParachainBackgroundTaskAfterSubscription { /// Stream of blocks of the relay chain this parachain is registered on. /// The buffer size should be large enough so that, if the CPU is busy, it doesn't become full /// before the execution of the sync service resumes. /// The maximum number of pinned block is ignored, as this maximum is a way to avoid malicious /// behaviors. This code is by definition not considered malicious. - relay_chain_subscribe_all: runtime_service::Subscription, - - /// Hash of the best parachain that has been reported to the subscriptions. - /// `None` if and only if no finalized parachain head is known yet. - reported_best_parahead_hash: Option<[u8; 32]>, - - /// Tree of relay chain blocks. Blocks are inserted when received from the relay chain - /// sync service. Once inside, their corresponding parachain head is fetched. Once the - /// parachain head is fetched, this parachain head is reported to our subscriptions. - /// - /// The root of the tree is a "virtual" block. It can be thought as the parent of the relay - /// chain finalized block, but is there even if the relay chain finalized block is block 0. - /// - /// All block in the tree has an associated parachain head behind an `Option`. This `Option` - /// always contains `Some`, except for the "virtual" root block for which it is `None`. - /// - /// If the output finalized block has a parachain head equal to `None`, it therefore means - /// that no finalized parachain head is known yet. - /// Note that, when it is the case, `SubscribeAll` messages from the frontend are still - /// answered with a single finalized block set to `obsolete_finalized_parahead`. Once a - /// finalized parachain head is known, it is important to reset all subscriptions. - /// - /// The set of blocks in this tree whose parachain block hasn't been fetched yet is the same - /// as the set of blocks that is maintained pinned on the runtime service. Blocks are unpinned - /// when their parachain head fetching succeeds or when they are removed from the tree. - async_tree: async_tree::AsyncTree>>, + paraheads_subscribe_all: async_channel::Receiver, - /// If `true`, [`ParachainBackgroundTaskAfterSubscription::async_tree`] might need to - /// be advanced. - must_process_sync_tree: bool, - - /// List of in-progress parachain head fetching operations. - /// - /// The operations require some blocks to be pinned within the relay chain runtime service, - /// which is guaranteed by the fact that `relay_chain_subscribe_all.new_blocks` stays - /// alive for longer than this container, and by the fact that we unpin block after a - /// fetching operation has finished and that we never fetch twice for the same block. - in_progress_paraheads: stream::FuturesUnordered< - future::BoxFuture<'static, (async_tree::AsyncOpId, Result, ParaheadError>)>, - >, - - /// Future that is ready when we need to start a new parachain head fetch operation. - next_start_parahead_fetch: Pin + Send>>, + /// List of block numbers indexed by the block hash for all non-finalized blocks. Filled + /// when we receive a notification from the paraheads task. + /// Blocks whose header isn't decodable are ignored. + known_block_numbers: HashMap<[u8; 32], u64, fnv::FnvBuildHasher>, } impl ParachainBackgroundTask { @@ -200,43 +167,28 @@ impl ParachainBackgroundTask { futures_lite::future::yield_now().await; // Wait until something interesting happens. - enum WakeUpReason { + enum WakeUpReason { ForegroundClosed, ForegroundMessage(ToBackground), - NewSubscription(runtime_service::SubscribeAll), - StartParaheadFetch, - ParaheadFetchFinished { - async_op_id: async_tree::AsyncOpId, - parahead_result: Result, ParaheadError>, - }, - Notification(runtime_service::Notification), + NewSubscription(super::SubscribeAll), + ParaheadNotification(super::Notification), SubscriptionDead, MustSubscribeNetworkEvents, NetworkEvent(network_service::Event), - AdvanceSyncTree, } - let wake_up_reason: WakeUpReason<_> = { - let ( - subscribe_future, - next_start_parahead_fetch, - relay_chain_subscribe_all, - in_progress_paraheads, - must_process_sync_tree, - is_relaychain_subscribed, - ) = match &mut self.subscription_state { - ParachainBackgroundState::NotSubscribed { - subscribe_future, .. - } => (Some(subscribe_future), None, None, None, None, false), - ParachainBackgroundState::Subscribed(runtime_subscription) => ( - None, - Some(&mut runtime_subscription.next_start_parahead_fetch), - Some(&mut runtime_subscription.relay_chain_subscribe_all), - Some(&mut runtime_subscription.in_progress_paraheads), - Some(&mut runtime_subscription.must_process_sync_tree), - true, - ), - }; + let wake_up_reason: WakeUpReason = { + let (subscribe_future, paraheads_subscribe_all, is_paraheads_subscribed) = + match &mut self.subscription_state { + ParachainBackgroundState::NotSubscribed { + subscribe_future, .. + } => (Some(subscribe_future), None, false), + ParachainBackgroundState::Subscribed(paraheads_subscription) => ( + None, + Some(&mut paraheads_subscription.paraheads_subscribe_all), + true, + ), + }; async { if let Some(subscribe_future) = subscribe_future { @@ -252,17 +204,17 @@ impl ParachainBackgroundTask { } }) .or(async { - if let Some(relay_chain_subscribe_all) = relay_chain_subscribe_all { - match relay_chain_subscribe_all.next().await { - Some(notif) => WakeUpReason::Notification(notif), - None => WakeUpReason::SubscriptionDead, + if let Some(paraheads_subscribe_all) = paraheads_subscribe_all { + match paraheads_subscribe_all.recv().await { + Ok(notif) => WakeUpReason::ParaheadNotification(notif), + Err(_) => WakeUpReason::SubscriptionDead, } } else { future::pending().await } }) .or(async { - if is_relaychain_subscribed { + if is_paraheads_subscribed { if let Some(from_network_service) = self.from_network_service.as_mut() { match from_network_service.next().await { Some(ev) => WakeUpReason::NetworkEvent(ev), @@ -278,43 +230,6 @@ impl ParachainBackgroundTask { future::pending().await } }) - .or(async { - if let Some(next_start_parahead_fetch) = next_start_parahead_fetch { - next_start_parahead_fetch.as_mut().await; - *next_start_parahead_fetch = Box::pin(future::pending()); - WakeUpReason::StartParaheadFetch - } else { - future::pending().await - } - }) - .or(async { - if let Some(in_progress_paraheads) = in_progress_paraheads { - if !in_progress_paraheads.is_empty() { - let (async_op_id, parahead_result) = - in_progress_paraheads.next().await.unwrap(); - WakeUpReason::ParaheadFetchFinished { - async_op_id, - parahead_result, - } - } else { - future::pending().await - } - } else { - future::pending().await - } - }) - .or(async { - if let Some(must_process_sync_tree) = must_process_sync_tree { - if *must_process_sync_tree { - *must_process_sync_tree = false; - WakeUpReason::AdvanceSyncTree - } else { - future::pending().await - } - } else { - future::pending().await - } - }) .await }; @@ -324,710 +239,132 @@ impl ParachainBackgroundTask { return; } - (WakeUpReason::NewSubscription(relay_chain_subscribe_all), _) => { - // Subscription to the relay chain has finished. + (WakeUpReason::NewSubscription(paraheads_subscribe_all), _) => { + // Subscription to the paraheads has finished. log!( &self.platform, Debug, &self.log_target, - "relay-chain-new-subscription", + "paraheads-new-subscription", finalized_hash = HashDisplay(&header::hash_from_scale_encoded_header( - &relay_chain_subscribe_all.finalized_block_scale_encoded_header + ¶heads_subscribe_all.finalized_block_scale_encoded_header )), - subscription_id = ?relay_chain_subscribe_all.new_blocks.id(), - ); - log!( - &self.platform, - Debug, - &self.log_target, - "parahead-fetch-operations-cleared" ); - let async_tree = { - let mut async_tree = - async_tree::AsyncTree::::new( - async_tree::Config { - finalized_async_user_data: None, - retry_after_failed: Duration::from_secs(5), - blocks_capacity: 32, - }, - ); - let finalized_hash = header::hash_from_scale_encoded_header( - &relay_chain_subscribe_all.finalized_block_scale_encoded_header, + // The networking service needs to be kept up to date with what the + // local node considers as the best block. + if let Ok(header) = header::decode( + ¶heads_subscribe_all.finalized_block_scale_encoded_header, + self.block_number_bytes, + ) { + let parahash = header::hash_from_scale_encoded_header( + ¶heads_subscribe_all.finalized_block_scale_encoded_header, ); - let finalized_index = - async_tree.input_insert_block(finalized_hash, None, false, true); - async_tree.input_finalize(finalized_index); - for block in relay_chain_subscribe_all.non_finalized_blocks_ancestry_order { - let hash = - header::hash_from_scale_encoded_header(&block.scale_encoded_header); - let parent = async_tree - .input_output_iter_unordered() - .find(|b| *b.user_data == block.parent_hash) - .map(|b| b.id) - .unwrap_or(finalized_index); - async_tree.input_insert_block( - hash, - Some(parent), - false, - block.is_new_best, - ); - } - async_tree - }; + self.network_service + .set_local_best_block(parahash, header.number) + .await; + } + + let known_block_numbers = paraheads_subscribe_all + .non_finalized_blocks_ancestry_order + .iter() + .filter_map(|block| { + Some(( + header::hash_from_scale_encoded_header(&block.scale_encoded_header), + header::decode( + &block.scale_encoded_header, + self.block_number_bytes, + ) + .ok()? + .number, + )) + }) + .collect(); self.subscription_state = ParachainBackgroundState::Subscribed( ParachainBackgroundTaskAfterSubscription { - all_subscriptions: match &mut self.subscription_state { - ParachainBackgroundState::NotSubscribed { - all_subscriptions, - .. - } => mem::take(all_subscriptions), - _ => unreachable!(), - }, - relay_chain_subscribe_all: relay_chain_subscribe_all.new_blocks, - reported_best_parahead_hash: None, - async_tree, - must_process_sync_tree: false, - in_progress_paraheads: stream::FuturesUnordered::new(), - next_start_parahead_fetch: Box::pin(future::ready(())), + known_block_numbers, + paraheads_subscribe_all: paraheads_subscribe_all.new_blocks, }, ); } - ( - WakeUpReason::AdvanceSyncTree, - ParachainBackgroundState::Subscribed(runtime_subscription), - ) => { - if let Some(update) = runtime_subscription.async_tree.try_advance_output() { - // Make sure to process any notification that comes after. - runtime_subscription.must_process_sync_tree = true; - - match update { - async_tree::OutputUpdate::Finalized { - former_finalized_async_op_user_data: former_finalized_parahead, - pruned_blocks, - best_output_block_updated, - .. - } if *runtime_subscription - .async_tree - .output_finalized_async_user_data() - != former_finalized_parahead => - { - let new_finalized_parahead = runtime_subscription - .async_tree - .output_finalized_async_user_data(); - debug_assert!(new_finalized_parahead.is_some()); - - // If this is the first time a finalized parahead is known, any - // `SubscribeAll` message that has been answered beforehand was - // answered in a dummy way with a potentially obsolete finalized - // header. - // For this reason, we reset all subscriptions to force all - // subscribers to re-subscribe. - if former_finalized_parahead.is_none() { - runtime_subscription.all_subscriptions.clear(); - } - - let hash = header::hash_from_scale_encoded_header( - new_finalized_parahead.as_ref().unwrap(), - ); - - self.obsolete_finalized_parahead = - new_finalized_parahead.clone().unwrap(); - - // Must unpin the pruned blocks if they haven't already been unpinned. - let mut pruned_blocks_hashes = - Vec::with_capacity(pruned_blocks.len()); - for (_, hash, pruned_block_parahead) in pruned_blocks { - if pruned_block_parahead.is_none() { - runtime_subscription - .relay_chain_subscribe_all - .unpin_block(hash) - .await; - } - pruned_blocks_hashes.push(hash); - } - - log!( - &self.platform, - Debug, - &self.log_target, - "subscriptions-notify-parablock-finalized", - hash = HashDisplay(&hash) - ); - - let best_block_hash = runtime_subscription - .async_tree - .output_best_block_index() - .map(|(_, parahead)| { - header::hash_from_scale_encoded_header( - parahead.as_ref().unwrap(), - ) - }) - .unwrap_or(hash); - runtime_subscription.reported_best_parahead_hash = - Some(best_block_hash); - - // Elements in `all_subscriptions` are removed one by one and - // inserted back if the channel is still open. - for index in (0..runtime_subscription.all_subscriptions.len()).rev() - { - let sender = - runtime_subscription.all_subscriptions.swap_remove(index); - let notif = super::Notification::Finalized { - hash, - best_block_hash_if_changed: if best_output_block_updated { - Some(best_block_hash) - } else { - None - }, - pruned_blocks: pruned_blocks_hashes.clone(), - }; - if sender.try_send(notif).is_ok() { - runtime_subscription.all_subscriptions.push(sender); - } - } - } - - async_tree::OutputUpdate::Finalized { .. } - | async_tree::OutputUpdate::BestBlockChanged { .. } => { - // Do not report anything to subscriptions if no finalized parahead is - // known yet. - let finalized_parahead = match runtime_subscription - .async_tree - .output_finalized_async_user_data() - { - Some(p) => p, - None => continue, - }; - - // Calculate hash of the parablock corresponding to the new best relay - // chain block. - let parahash = header::hash_from_scale_encoded_header( - runtime_subscription - .async_tree - .output_best_block_index() - .map(|(_, b)| b.as_ref().unwrap()) - .unwrap_or(finalized_parahead), - ); - - if runtime_subscription.reported_best_parahead_hash.as_ref() - != Some(¶hash) - { - runtime_subscription.reported_best_parahead_hash = - Some(parahash); - - // The networking service needs to be kept up to date with what the local - // node considers as the best block. - if let Ok(header) = - header::decode(finalized_parahead, self.block_number_bytes) - { - self.network_service - .set_local_best_block(parahash, header.number) - .await; - } - - log!( - &self.platform, - Debug, - &self.log_target, - "subscriptions-notify-best-block-changed", - hash = HashDisplay(¶hash) - ); - - // Elements in `all_subscriptions` are removed one by one and - // inserted back if the channel is still open. - for index in - (0..runtime_subscription.all_subscriptions.len()).rev() - { - let sender = runtime_subscription - .all_subscriptions - .swap_remove(index); - let notif = super::Notification::BestBlockChanged { - hash: parahash, - }; - if sender.try_send(notif).is_ok() { - runtime_subscription.all_subscriptions.push(sender); - } - } - } - } - - async_tree::OutputUpdate::Block(block) => { - // `block` borrows `async_tree`. We need to mutably access `async_tree` - // below, so deconstruct `block` beforehand. - let is_new_best = block.is_new_best; - let block_index = block.index; - let scale_encoded_header: Vec = runtime_subscription - .async_tree - .block_async_user_data(block.index) - .unwrap() - .clone() - .unwrap(); - let parahash = - header::hash_from_scale_encoded_header(&scale_encoded_header); - - // Do not report anything to subscriptions if no finalized parahead is - // known yet. - let finalized_parahead = match runtime_subscription - .async_tree - .output_finalized_async_user_data() - { - Some(p) => p, - None => continue, - }; - - // Do not report the new block if it has already been reported in the - // past. This covers situations where the parahead is identical to the - // relay chain's parent's parahead, but also situations where multiple - // sibling relay chain blocks have the same parahead. - if *finalized_parahead == scale_encoded_header - || runtime_subscription - .async_tree - .input_output_iter_unordered() - .filter(|item| item.id != block_index) - .filter_map(|item| item.async_op_user_data) - .any(|item| item.as_ref() == Some(&scale_encoded_header)) - { - // While the parablock has already been reported, it is possible that - // it becomes the new best block while it wasn't before, in which - // case we should send a notification. - if is_new_best - && runtime_subscription.reported_best_parahead_hash.as_ref() - != Some(¶hash) - { - runtime_subscription.reported_best_parahead_hash = - Some(parahash); - - // The networking service needs to be kept up to date with what the - // local node considers as the best block. - if let Ok(header) = header::decode( - finalized_parahead, - self.block_number_bytes, - ) { - self.network_service - .set_local_best_block(parahash, header.number) - .await; - } - - log!( - &self.platform, - Debug, - &self.log_target, - "subscriptions-notify-best-block-changed", - hash = HashDisplay(¶hash) - ); - - // Elements in `all_subscriptions` are removed one by one and - // inserted back if the channel is still open. - for index in - (0..runtime_subscription.all_subscriptions.len()).rev() - { - let sender = runtime_subscription - .all_subscriptions - .swap_remove(index); - let notif = super::Notification::BestBlockChanged { - hash: parahash, - }; - if sender.try_send(notif).is_ok() { - runtime_subscription.all_subscriptions.push(sender); - } - } - } - - continue; - } - - if is_new_best { - runtime_subscription.reported_best_parahead_hash = - Some(parahash); - } - - let parent_hash = header::hash_from_scale_encoded_header( - runtime_subscription - .async_tree - .parent(block_index) - .map(|idx| { - runtime_subscription - .async_tree - .block_async_user_data(idx) - .unwrap() - .as_ref() - .unwrap() - }) - .unwrap_or(finalized_parahead), - ); - - log!( - &self.platform, - Debug, - &self.log_target, - "subscriptions-notify-new-parablock", - hash = HashDisplay(¶hash), - parent_hash = HashDisplay(&parent_hash), - ?is_new_best - ); - - // Elements in `all_subscriptions` are removed one by one and - // inserted back if the channel is still open. - for index in (0..runtime_subscription.all_subscriptions.len()).rev() - { - let sender = - runtime_subscription.all_subscriptions.swap_remove(index); - let notif = - super::Notification::Block(super::BlockNotification { - is_new_best, - parent_hash, - scale_encoded_header: scale_encoded_header.clone(), - }); - if sender.try_send(notif).is_ok() { - runtime_subscription.all_subscriptions.push(sender); - } - } - } - } - } - } - - ( - WakeUpReason::StartParaheadFetch, - ParachainBackgroundState::Subscribed(runtime_subscription), - ) => { - // Must start downloading a parahead. - - // Internal state check. - debug_assert_eq!( - runtime_subscription.reported_best_parahead_hash.is_some(), - runtime_subscription - .async_tree - .output_finalized_async_user_data() - .is_some() - ); - - // Limit the maximum number of simultaneous downloads. - if runtime_subscription.in_progress_paraheads.len() >= 4 { - continue; - } - - match runtime_subscription - .async_tree - .next_necessary_async_op(&self.platform.now()) - { - async_tree::NextNecessaryAsyncOp::NotReady { when: Some(when) } => { - runtime_subscription.next_start_parahead_fetch = - Box::pin(self.platform.sleep_until(when)); - } - async_tree::NextNecessaryAsyncOp::NotReady { when: None } => { - runtime_subscription.next_start_parahead_fetch = - Box::pin(future::pending()); - } - async_tree::NextNecessaryAsyncOp::Ready(op) => { - log!( - &self.platform, - Debug, - &self.log_target, - "parahead-fetch-operation-started", - relay_block_hash = - HashDisplay(&runtime_subscription.async_tree[op.block_index]), - ); - - runtime_subscription.in_progress_paraheads.push({ - let relay_chain_sync = self.relay_chain_sync.clone(); - let subscription_id = - runtime_subscription.relay_chain_subscribe_all.id(); - let block_hash = runtime_subscription.async_tree[op.block_index]; - let async_op_id = op.id; - let parachain_id = self.parachain_id; - Box::pin(async move { - ( - async_op_id, - fetch_parahead( - &relay_chain_sync, - subscription_id, - parachain_id, - &block_hash, - ) - .await, - ) - }) - }); - - // There might be more downloads to start. - runtime_subscription.next_start_parahead_fetch = - Box::pin(future::ready(())); - } - } - } - - ( - WakeUpReason::Notification(runtime_service::Notification::Finalized { - hash, - best_block_hash_if_changed, - .. - }), - ParachainBackgroundState::Subscribed(runtime_subscription), - ) => { - // Relay chain has a new finalized block. - log!( - &self.platform, - Debug, - &self.log_target, - "relay-chain-block-finalized", - hash = HashDisplay(&hash) - ); - - if let Some(best_block_hash_if_changed) = best_block_hash_if_changed { - let best = runtime_subscription - .async_tree - .input_output_iter_unordered() - .find(|b| *b.user_data == best_block_hash_if_changed) - .unwrap() - .id; - runtime_subscription - .async_tree - .input_set_best_block(Some(best)); - } - - let finalized = runtime_subscription - .async_tree - .input_output_iter_unordered() - .find(|b| *b.user_data == hash) - .unwrap() - .id; - runtime_subscription.async_tree.input_finalize(finalized); - runtime_subscription.must_process_sync_tree = true; - } - - ( - WakeUpReason::Notification(runtime_service::Notification::Block(block)), - ParachainBackgroundState::Subscribed(runtime_subscription), - ) => { - // Relay chain has a new block. - let hash = header::hash_from_scale_encoded_header(&block.scale_encoded_header); - - log!( - &self.platform, - Debug, - &self.log_target, - "relay-chain-new-block", - hash = HashDisplay(&hash), - parent_hash = HashDisplay(&block.parent_hash) - ); - - let parent = runtime_subscription - .async_tree - .input_output_iter_unordered() - .find(|b| *b.user_data == block.parent_hash) - .map(|b| b.id); // TODO: check if finalized - runtime_subscription.async_tree.input_insert_block( - hash, - parent, - false, - block.is_new_best, - ); - runtime_subscription.must_process_sync_tree = true; - - runtime_subscription.next_start_parahead_fetch = Box::pin(future::ready(())); - } - - ( - WakeUpReason::Notification(runtime_service::Notification::BestBlockChanged { - hash, - }), - ParachainBackgroundState::Subscribed(runtime_subscription), - ) => { - // Relay chain has a new best block. - log!( - &self.platform, - Debug, - &self.log_target, - "relay-chain-best-block-changed", - hash = HashDisplay(&hash) - ); - - // If the block isn't found in `async_tree`, assume that it is equal to the - // finalized block (that has left the tree already). - let node_idx = runtime_subscription - .async_tree - .input_output_iter_unordered() - .find(|b| *b.user_data == hash) - .map(|b| b.id); - runtime_subscription - .async_tree - .input_set_best_block(node_idx); - - runtime_subscription.must_process_sync_tree = true; - } - (WakeUpReason::SubscriptionDead, _) => { // Recreate the channel. log!( &self.platform, Debug, &self.log_target, - "relay-chain-subscription-reset" + "paraheads-subscription-reset" ); self.subscription_state = ParachainBackgroundState::NotSubscribed { - all_subscriptions: Vec::new(), subscribe_future: { - let relay_chain_sync = self.relay_chain_sync.clone(); + let to_paraheads = self.paraheads.clone(); Box::pin(async move { - relay_chain_sync - .subscribe_all(32, NonZero::::new(usize::MAX).unwrap()) - .await + let (send_back, sub_rx) = oneshot::channel(); + let _ = to_paraheads + .send(super::ToBackground::SubscribeAll { + send_back, + buffer_size: 32, + runtime_interest: false, + }) + .await; + // TODO: don't panic if the paraheads service dies here + sub_rx.await.unwrap() }) }, }; } ( - WakeUpReason::ParaheadFetchFinished { - async_op_id, - parahead_result: Ok(parahead), - }, - ParachainBackgroundState::Subscribed(runtime_subscription), + WakeUpReason::ParaheadNotification(super::Notification::Block( + super::BlockNotification { + scale_encoded_header, + .. + }, + )), + ParachainBackgroundState::Subscribed(paraheads_subscription), ) => { - // A parahead fetching operation is successful. - log!( - &self.platform, - Debug, - &self.log_target, - "parahead-fetch-operation-success", - parahead_hash = HashDisplay( - blake2_rfc::blake2b::blake2b(32, b"", ¶head).as_bytes() - ), - relay_blocks = runtime_subscription - .async_tree - .async_op_blocks(async_op_id) - .map(|b| HashDisplay(b)) - .join(",") - ); - - // Unpin the relay blocks whose parahead is now known. - for block in runtime_subscription - .async_tree - .async_op_finished(async_op_id, Some(parahead)) + if let Ok(header) = + header::decode(&scale_encoded_header, self.block_number_bytes) { - let hash = &runtime_subscription.async_tree[block]; - runtime_subscription - .relay_chain_subscribe_all - .unpin_block(*hash) - .await; + paraheads_subscription.known_block_numbers.insert( + header::hash_from_scale_encoded_header(&scale_encoded_header), + header.number, + ); } - - runtime_subscription.must_process_sync_tree = true; - - runtime_subscription.next_start_parahead_fetch = Box::pin(future::ready(())); } ( - WakeUpReason::ParaheadFetchFinished { - parahead_result: - Err(ParaheadError::PinRuntimeError( - runtime_service::PinPinnedBlockRuntimeError::ObsoleteSubscription, - )), + WakeUpReason::ParaheadNotification(super::Notification::Finalized { + hash, + pruned_blocks, .. - }, - _, - ) => { - // The relay chain runtime service has some kind of gap or issue and has - // discarded the runtime. - // Destroy the subscription and recreate the channels. - log!( - &self.platform, - Debug, - &self.log_target, - "relay-chain-subscription-reset" - ); - self.subscription_state = ParachainBackgroundState::NotSubscribed { - all_subscriptions: Vec::new(), - subscribe_future: { - let relay_chain_sync = self.relay_chain_sync.clone(); - Box::pin(async move { - relay_chain_sync - .subscribe_all(32, NonZero::::new(usize::MAX).unwrap()) - .await - }) - }, - }; - } - - ( - WakeUpReason::ParaheadFetchFinished { - async_op_id, - parahead_result: Err(error), - }, - ParachainBackgroundState::Subscribed(runtime_subscription), + }), + ParachainBackgroundState::Subscribed(paraheads_subscription), ) => { - // Failed fetching a parahead. - - // Several relay chains initially didn't support parachains, and have later - // been upgraded to support them. Similarly, the parachain might not have had a - // core on the relay chain until recently. For these reasons, errors when the - // relay chain is not near head of the chain are most likely normal and do - // not warrant logging an error. - // Note that `is_near_head_of_chain_heuristic` is normally not acceptable to - // use due to being too vague, but since this is just about whether to print a - // log message, it's completely fine. - if self - .relay_chain_sync - .is_near_head_of_chain_heuristic() - .await - && !error.is_network_problem() + // The networking service needs to be kept up to date with what the + // local node considers as the best block. + if let Some(block_number) = + paraheads_subscription.known_block_numbers.get(&hash) { - log!( - &self.platform, - Error, - &self.log_target, - format!( - "Failed to fetch the parachain head from relay chain blocks {}: {}", - runtime_subscription - .async_tree - .async_op_blocks(async_op_id) - .map(|b| HashDisplay(b)) - .join(", "), - error - ) - ); + self.network_service + .set_local_best_block(hash, *block_number) + .await; } - log!( - &self.platform, - Debug, - &self.log_target, - "parahead-fetch-operation-error", - relay_blocks = runtime_subscription - .async_tree - .async_op_blocks(async_op_id) - .map(|b| HashDisplay(b)) - .join(","), - ?error - ); - - runtime_subscription - .async_tree - .async_op_failure(async_op_id, &self.platform.now()); - - runtime_subscription.next_start_parahead_fetch = Box::pin(future::ready(())); + for pruned in iter::chain([hash], pruned_blocks) { + paraheads_subscription.known_block_numbers.remove(&pruned); + } } ( - WakeUpReason::ForegroundMessage(ToBackground::IsNearHeadOfChainHeuristic { - send_back, + WakeUpReason::ParaheadNotification(super::Notification::BestBlockChanged { + .. }), - ParachainBackgroundState::Subscribed(sub), - ) if sub.async_tree.output_finalized_async_user_data().is_some() => { - // Since there is a mapping between relay chain blocks and parachain blocks, - // whether a parachain is at the head of the chain is the same thing as whether - // its relay chain is at the head of the chain. - // Note that there is no ordering guarantee of any kind w.r.t. block - // subscriptions notifications. - let val = self - .relay_chain_sync - .is_near_head_of_chain_heuristic() - .await; - let _ = send_back.send(val); + ParachainBackgroundState::Subscribed(_), + ) => { + // Ignored. } ( @@ -1036,164 +373,28 @@ impl ParachainBackgroundTask { }), _, ) => { - // If no finalized parahead is known yet, we might be very close to the head - // but also maybe very very far away. We lean on the cautious side and always - // return `false`. - let _ = send_back.send(false); + let _ = self + .paraheads + .send(ToBackground::IsNearHeadOfChainHeuristic { send_back }) + .await; } ( WakeUpReason::ForegroundMessage(ToBackground::SubscribeAll { send_back, buffer_size, - .. + runtime_interest, }), - ParachainBackgroundState::NotSubscribed { - all_subscriptions, .. - }, - ) => { - let (tx, new_blocks) = async_channel::bounded(buffer_size.saturating_sub(1)); - - // No known finalized parahead. - let _ = send_back.send(super::SubscribeAll { - finalized_block_scale_encoded_header: self - .obsolete_finalized_parahead - .clone(), - finalized_block_runtime: None, - non_finalized_blocks_ancestry_order: Vec::new(), - new_blocks, - }); - - all_subscriptions.push(tx); - } - - ( - WakeUpReason::ForegroundMessage(ToBackground::SubscribeAll { - send_back, - buffer_size, - .. - }), - ParachainBackgroundState::Subscribed(runtime_subscription), + _, ) => { - let (tx, new_blocks) = async_channel::bounded(buffer_size.saturating_sub(1)); - - // There are two possibilities here: either we know of any recent finalized - // parahead, or we don't. In case where we don't know of any finalized parahead - // yet, we report a single obsolete finalized parahead, which is - // `obsolete_finalized_parahead`. The rest of this module makes sure that no - // other block is reported to subscriptions as long as this is the case, and - // that subscriptions are reset once the first known finalized parahead - // is known. - if let Some(finalized_parahead) = runtime_subscription - .async_tree - .output_finalized_async_user_data() - { - // Finalized parahead is known. - let finalized_parahash = - header::hash_from_scale_encoded_header(finalized_parahead); - let _ = send_back.send(super::SubscribeAll { - finalized_block_scale_encoded_header: finalized_parahead.clone(), - finalized_block_runtime: None, - non_finalized_blocks_ancestry_order: { - let mut list = - Vec::<([u8; 32], super::BlockNotification)>::with_capacity( - runtime_subscription - .async_tree - .num_input_non_finalized_blocks(), - ); - - for relay_block in runtime_subscription - .async_tree - .input_output_iter_ancestry_order() - { - let parablock = match relay_block.async_op_user_data { - Some(b) => b.as_ref().unwrap(), - None => continue, - }; - - let parablock_hash = - header::hash_from_scale_encoded_header(parablock); - - // TODO: O(n) - if let Some((_, entry)) = - list.iter_mut().find(|(h, _)| *h == parablock_hash) - { - // Block is already in the list. Don't add it a second time. - if relay_block.is_output_best { - entry.is_new_best = true; - } - continue; - } - - // Find the parent of the parablock. This is done by going through - // the ancestors of the corresponding relay chain block (until and - // including the finalized relay chain block) until we find one - // whose parablock is different from the parablock in question. - // If none is found, the parablock is the same as the finalized - // parablock. - let parent_hash = runtime_subscription - .async_tree - .ancestors(relay_block.id) - .find_map(|idx| { - let hash = header::hash_from_scale_encoded_header( - runtime_subscription - .async_tree - .block_async_user_data(idx) - .unwrap() - .as_ref() - .unwrap(), - ); - if hash != parablock_hash { - Some(hash) - } else { - None - } - }) - .or_else(|| { - if finalized_parahash != parablock_hash { - Some(finalized_parahash) - } else { - None - } - }); - - // `parent_hash` is `None` if the parablock is - // the same as the finalized parablock, in which case we - // don't add it to the list. - if let Some(parent_hash) = parent_hash { - debug_assert!( - list.iter().filter(|(h, _)| *h == parent_hash).count() - == 1 - || parent_hash == finalized_parahash - ); - list.push(( - parablock_hash, - super::BlockNotification { - is_new_best: relay_block.is_output_best, - scale_encoded_header: parablock.clone(), - parent_hash, - }, - )); - } - } - - list.into_iter().map(|(_, v)| v).collect() - }, - new_blocks, - }); - } else { - // No known finalized parahead. - let _ = send_back.send(super::SubscribeAll { - finalized_block_scale_encoded_header: self - .obsolete_finalized_parahead - .clone(), - finalized_block_runtime: None, - non_finalized_blocks_ancestry_order: Vec::new(), - new_blocks, - }); - } - - runtime_subscription.all_subscriptions.push(tx); + let _ = self + .paraheads + .send(ToBackground::SubscribeAll { + send_back, + buffer_size, + runtime_interest, + }) + .await; } ( @@ -1233,7 +434,10 @@ impl ParachainBackgroundTask { }), _, ) => { - let _ = send_back.send(None); + let _ = self + .paraheads + .send(ToBackground::SerializeChainInformation { send_back }) + .await; } (WakeUpReason::MustSubscribeNetworkEvents, _) => { @@ -1283,10 +487,7 @@ impl ParachainBackgroundTask { } ( - WakeUpReason::ParaheadFetchFinished { .. } - | WakeUpReason::AdvanceSyncTree - | WakeUpReason::Notification(_) - | WakeUpReason::StartParaheadFetch, + WakeUpReason::ParaheadNotification(_), ParachainBackgroundState::NotSubscribed { .. }, ) => { // These paths are unreachable. @@ -1296,85 +497,3 @@ impl ParachainBackgroundTask { } } } - -async fn fetch_parahead( - relay_chain_sync: &Arc>, - subscription_id: runtime_service::SubscriptionId, - parachain_id: u32, - block_hash: &[u8; 32], -) -> Result, ParaheadError> { - // Call `ParachainHost_persisted_validation_data` in order to know where the parachain is. - let (pinned_runtime, block_state_trie_root, block_number) = relay_chain_sync - .pin_pinned_block_runtime(subscription_id, *block_hash) - .await - .map_err(ParaheadError::PinRuntimeError)?; - let success = relay_chain_sync - .runtime_call( - pinned_runtime, - *block_hash, - block_number, - block_state_trie_root, - para::PERSISTED_VALIDATION_FUNCTION_NAME.to_owned(), - None, // TODO: /!\ - para::persisted_validation_data_parameters( - parachain_id, - para::OccupiedCoreAssumption::TimedOut, - ) - .fold(Vec::new(), |mut a, b| { - a.extend_from_slice(b.as_ref()); - a - }), - 6, - Duration::from_secs(10), - NonZero::::new(2).unwrap(), - ) - .await - .map_err(ParaheadError::RuntimeCall)?; - - // Try decode the result of the runtime call. - // If this fails, it indicates an incompatibility between smoldot and the relay chain. - match para::decode_persisted_validation_data_return_value( - &success.output, - relay_chain_sync.block_number_bytes(), - ) { - Ok(Some(pvd)) => Ok(pvd.parent_head.to_vec()), - Ok(None) => Err(ParaheadError::NoCore), - Err(error) => Err(ParaheadError::InvalidRuntimeOutput(error)), - } -} - -/// Error that can happen when fetching the parachain head corresponding to a relay chain block. -#[derive(Debug, derive_more::Display, derive_more::Error)] -enum ParaheadError { - /// Error while performing call request over the network. - #[display("Error while performing call request over the network: {_0}")] - RuntimeCall(runtime_service::RuntimeCallError), - /// Error pinning the runtime of the block. - PinRuntimeError(runtime_service::PinPinnedBlockRuntimeError), - /// Parachain doesn't have a core in the relay chain. - NoCore, - /// Error while decoding the output of the call. - /// - /// This indicates some kind of incompatibility between smoldot and the relay chain. - #[display("Error while decoding the output of the call: {_0}")] - InvalidRuntimeOutput(para::Error), -} - -impl ParaheadError { - /// Returns `true` if this is caused by networking issues, as opposed to a consensus-related - /// issue. - fn is_network_problem(&self) -> bool { - match self { - ParaheadError::RuntimeCall(runtime_service::RuntimeCallError::Inaccessible(_)) => true, - ParaheadError::RuntimeCall( - runtime_service::RuntimeCallError::Execution(_) - | runtime_service::RuntimeCallError::Crash - | runtime_service::RuntimeCallError::InvalidRuntime(_) - | runtime_service::RuntimeCallError::ApiVersionRequirementUnfulfilled, - ) => false, - ParaheadError::PinRuntimeError(_) => false, - ParaheadError::NoCore => false, - ParaheadError::InvalidRuntimeOutput(_) => false, - } - } -} diff --git a/light-base/src/sync_service/paraheads.rs b/light-base/src/sync_service/paraheads.rs new file mode 100644 index 0000000000..c1e1d9f410 --- /dev/null +++ b/light-base/src/sync_service/paraheads.rs @@ -0,0 +1,1235 @@ +// Smoldot +// Copyright (C) 2019-2022 Parity Technologies (UK) Ltd. +// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0 + +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. + +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU General Public License for more details. + +// You should have received a copy of the GNU General Public License +// along with this program. If not, see . + +use super::ToBackground; +use crate::{log, platform::PlatformRef, runtime_service}; + +use alloc::{borrow::ToOwned as _, boxed::Box, format, string::String, sync::Arc, vec::Vec}; +use core::{mem, num::NonZero, pin::Pin, time::Duration}; +use futures_lite::FutureExt as _; +use futures_util::{StreamExt as _, future, stream}; +use itertools::Itertools as _; +use smoldot::{chain::async_tree, header, informant::HashDisplay, sync::para}; + +/// Starts a sync service background task to synchronize a parachain. +pub(super) async fn start_paraheads( + log_target: String, + platform: TPlat, + finalized_block_header: Vec, + relay_chain_sync: Arc>, + parachain_id: u32, + from_foreground: Pin>>, +) { + ParachainBackgroundTask { + log_target, + from_foreground, + parachain_id, + obsolete_finalized_parahead: finalized_block_header, + subscription_state: ParachainBackgroundState::NotSubscribed { + all_subscriptions: Vec::new(), + subscribe_future: { + let relay_chain_sync = relay_chain_sync.clone(); + Box::pin(async move { + relay_chain_sync + .subscribe_all(32, NonZero::::new(usize::MAX).unwrap()) + .await + }) + }, + }, + relay_chain_sync, + platform, + } + .run() + .await; +} + +/// Task that is running in the background. +struct ParachainBackgroundTask { + /// Target to use for all logs. + log_target: String, + + /// Access to the platform's capabilities. + platform: TPlat, + + /// Channel receiving message from the sync service frontend. + from_foreground: Pin>>, + + /// Id of the parachain registered within the relay chain. Chosen by the user. + parachain_id: u32, + + /// Runtime service of the relay chain. + relay_chain_sync: Arc>, + + /// Last-known finalized parachain header. Can be very old and obsolete. + /// Updated after we successfully fetch the parachain head of a relay chain finalized block, + /// and left untouched if the fetch fails. + /// Initialized to the parachain genesis block header. + obsolete_finalized_parahead: Vec, + + /// Extra fields that are set after the subscription to the runtime service events has + /// succeeded. + subscription_state: ParachainBackgroundState, +} + +enum ParachainBackgroundState { + /// Currently subscribing to the relay chain runtime service. + NotSubscribed { + /// List of senders that will get notified when the tree of blocks is modified. + /// + /// These subscriptions are pending and no notification should be sent to them until the + /// subscription to the relay chain runtime service is finished. + all_subscriptions: Vec>, + + /// Future when the subscription has finished. + subscribe_future: future::BoxFuture<'static, runtime_service::SubscribeAll>, + }, + + /// Subscribed to the relay chain runtime service. + Subscribed(ParachainBackgroundTaskAfterSubscription), +} + +struct ParachainBackgroundTaskAfterSubscription { + /// List of senders that get notified when the tree of blocks is modified. + all_subscriptions: Vec>, + + /// Stream of blocks of the relay chain this parachain is registered on. + /// The buffer size should be large enough so that, if the CPU is busy, it doesn't become full + /// before the execution of the sync service resumes. + /// The maximum number of pinned block is ignored, as this maximum is a way to avoid malicious + /// behaviors. This code is by definition not considered malicious. + relay_chain_subscribe_all: runtime_service::Subscription, + + /// Hash of the best parachain that has been reported to the subscriptions. + /// `None` if and only if no finalized parachain head is known yet. + reported_best_parahead_hash: Option<[u8; 32]>, + + /// Tree of relay chain blocks. Blocks are inserted when received from the relay chain + /// sync service. Once inside, their corresponding parachain head is fetched. Once the + /// parachain head is fetched, this parachain head is reported to our subscriptions. + /// + /// The root of the tree is a "virtual" block. It can be thought as the parent of the relay + /// chain finalized block, but is there even if the relay chain finalized block is block 0. + /// + /// All block in the tree has an associated parachain head behind an `Option`. This `Option` + /// always contains `Some`, except for the "virtual" root block for which it is `None`. + /// + /// If the output finalized block has a parachain head equal to `None`, it therefore means + /// that no finalized parachain head is known yet. + /// Note that, when it is the case, `SubscribeAll` messages from the frontend are still + /// answered with a single finalized block set to `obsolete_finalized_parahead`. Once a + /// finalized parachain head is known, it is important to reset all subscriptions. + /// + /// The set of blocks in this tree whose parachain block hasn't been fetched yet is the same + /// as the set of blocks that is maintained pinned on the runtime service. Blocks are unpinned + /// when their parachain head fetching succeeds or when they are removed from the tree. + async_tree: async_tree::AsyncTree>>, + + /// If `true`, [`ParachainBackgroundTaskAfterSubscription::async_tree`] might need to + /// be advanced. + must_process_sync_tree: bool, + + /// List of in-progress parachain head fetching operations. + /// + /// The operations require some blocks to be pinned within the relay chain runtime service, + /// which is guaranteed by the fact that `relay_chain_subscribe_all.new_blocks` stays + /// alive for longer than this container, and by the fact that we unpin block after a + /// fetching operation has finished and that we never fetch twice for the same block. + in_progress_paraheads: stream::FuturesUnordered< + future::BoxFuture<'static, (async_tree::AsyncOpId, Result, ParaheadError>)>, + >, + + /// Future that is ready when we need to start a new parachain head fetch operation. + next_start_parahead_fetch: Pin + Send>>, +} + +impl ParachainBackgroundTask { + async fn run(mut self) { + loop { + // Yield at every loop in order to provide better tasks granularity. + futures_lite::future::yield_now().await; + + // Wait until something interesting happens. + enum WakeUpReason { + ForegroundClosed, + ForegroundMessage(ToBackground), + NewSubscription(runtime_service::SubscribeAll), + StartParaheadFetch, + ParaheadFetchFinished { + async_op_id: async_tree::AsyncOpId, + parahead_result: Result, ParaheadError>, + }, + Notification(runtime_service::Notification), + SubscriptionDead, + AdvanceSyncTree, + } + + let wake_up_reason: WakeUpReason<_> = { + let ( + subscribe_future, + next_start_parahead_fetch, + relay_chain_subscribe_all, + in_progress_paraheads, + must_process_sync_tree, + ) = match &mut self.subscription_state { + ParachainBackgroundState::NotSubscribed { + subscribe_future, .. + } => (Some(subscribe_future), None, None, None, None), + ParachainBackgroundState::Subscribed(runtime_subscription) => ( + None, + Some(&mut runtime_subscription.next_start_parahead_fetch), + Some(&mut runtime_subscription.relay_chain_subscribe_all), + Some(&mut runtime_subscription.in_progress_paraheads), + Some(&mut runtime_subscription.must_process_sync_tree), + ), + }; + + async { + if let Some(subscribe_future) = subscribe_future { + WakeUpReason::NewSubscription(subscribe_future.await) + } else { + future::pending().await + } + } + .or(async { + match self.from_foreground.next().await { + Some(msg) => WakeUpReason::ForegroundMessage(msg), + None => WakeUpReason::ForegroundClosed, + } + }) + .or(async { + if let Some(relay_chain_subscribe_all) = relay_chain_subscribe_all { + match relay_chain_subscribe_all.next().await { + Some(notif) => WakeUpReason::Notification(notif), + None => WakeUpReason::SubscriptionDead, + } + } else { + future::pending().await + } + }) + .or(async { + if let Some(next_start_parahead_fetch) = next_start_parahead_fetch { + next_start_parahead_fetch.as_mut().await; + *next_start_parahead_fetch = Box::pin(future::pending()); + WakeUpReason::StartParaheadFetch + } else { + future::pending().await + } + }) + .or(async { + if let Some(in_progress_paraheads) = in_progress_paraheads { + if !in_progress_paraheads.is_empty() { + let (async_op_id, parahead_result) = + in_progress_paraheads.next().await.unwrap(); + WakeUpReason::ParaheadFetchFinished { + async_op_id, + parahead_result, + } + } else { + future::pending().await + } + } else { + future::pending().await + } + }) + .or(async { + if let Some(must_process_sync_tree) = must_process_sync_tree { + if *must_process_sync_tree { + *must_process_sync_tree = false; + WakeUpReason::AdvanceSyncTree + } else { + future::pending().await + } + } else { + future::pending().await + } + }) + .await + }; + + match (wake_up_reason, &mut self.subscription_state) { + (WakeUpReason::ForegroundClosed, _) => { + // Terminate the background task. + return; + } + + (WakeUpReason::NewSubscription(relay_chain_subscribe_all), _) => { + // Subscription to the relay chain has finished. + log!( + &self.platform, + Debug, + &self.log_target, + "relay-chain-new-subscription", + finalized_hash = HashDisplay(&header::hash_from_scale_encoded_header( + &relay_chain_subscribe_all.finalized_block_scale_encoded_header + )), + subscription_id = ?relay_chain_subscribe_all.new_blocks.id(), + ); + log!( + &self.platform, + Debug, + &self.log_target, + "parahead-fetch-operations-cleared" + ); + + let async_tree = { + let mut async_tree = + async_tree::AsyncTree::::new( + async_tree::Config { + finalized_async_user_data: None, + retry_after_failed: Duration::from_secs(5), + blocks_capacity: 32, + }, + ); + let finalized_hash = header::hash_from_scale_encoded_header( + &relay_chain_subscribe_all.finalized_block_scale_encoded_header, + ); + let finalized_index = + async_tree.input_insert_block(finalized_hash, None, false, true); + async_tree.input_finalize(finalized_index); + for block in relay_chain_subscribe_all.non_finalized_blocks_ancestry_order { + let hash = + header::hash_from_scale_encoded_header(&block.scale_encoded_header); + let parent = async_tree + .input_output_iter_unordered() + .find(|b| *b.user_data == block.parent_hash) + .map(|b| b.id) + .unwrap_or(finalized_index); + async_tree.input_insert_block( + hash, + Some(parent), + false, + block.is_new_best, + ); + } + async_tree + }; + + self.subscription_state = ParachainBackgroundState::Subscribed( + ParachainBackgroundTaskAfterSubscription { + all_subscriptions: match &mut self.subscription_state { + ParachainBackgroundState::NotSubscribed { + all_subscriptions, + .. + } => mem::take(all_subscriptions), + _ => unreachable!(), + }, + relay_chain_subscribe_all: relay_chain_subscribe_all.new_blocks, + reported_best_parahead_hash: None, + async_tree, + must_process_sync_tree: false, + in_progress_paraheads: stream::FuturesUnordered::new(), + next_start_parahead_fetch: Box::pin(future::ready(())), + }, + ); + } + + ( + WakeUpReason::AdvanceSyncTree, + ParachainBackgroundState::Subscribed(runtime_subscription), + ) => { + if let Some(update) = runtime_subscription.async_tree.try_advance_output() { + // Make sure to process any notification that comes after. + runtime_subscription.must_process_sync_tree = true; + + match update { + async_tree::OutputUpdate::Finalized { + former_finalized_async_op_user_data: former_finalized_parahead, + pruned_blocks, + best_output_block_updated, + .. + } if *runtime_subscription + .async_tree + .output_finalized_async_user_data() + != former_finalized_parahead => + { + let new_finalized_parahead = runtime_subscription + .async_tree + .output_finalized_async_user_data(); + debug_assert!(new_finalized_parahead.is_some()); + + // If this is the first time a finalized parahead is known, any + // `SubscribeAll` message that has been answered beforehand was + // answered in a dummy way with a potentially obsolete finalized + // header. + // For this reason, we reset all subscriptions to force all + // subscribers to re-subscribe. + if former_finalized_parahead.is_none() { + runtime_subscription.all_subscriptions.clear(); + } + + let hash = header::hash_from_scale_encoded_header( + new_finalized_parahead.as_ref().unwrap(), + ); + + // Must unpin the pruned blocks if they haven't already been unpinned. + let mut pruned_blocks_hashes = + Vec::with_capacity(pruned_blocks.len()); + for (_, hash, pruned_block_parahead) in pruned_blocks { + if pruned_block_parahead.is_none() { + runtime_subscription + .relay_chain_subscribe_all + .unpin_block(hash) + .await; + } + pruned_blocks_hashes.push(hash); + } + + log!( + &self.platform, + Debug, + &self.log_target, + "subscriptions-notify-parablock-finalized", + hash = HashDisplay(&hash) + ); + + let best_block_hash = runtime_subscription + .async_tree + .output_best_block_index() + .map(|(_, parahead)| { + header::hash_from_scale_encoded_header( + parahead.as_ref().unwrap(), + ) + }) + .unwrap_or(hash); + runtime_subscription.reported_best_parahead_hash = + Some(best_block_hash); + + // Elements in `all_subscriptions` are removed one by one and + // inserted back if the channel is still open. + for index in (0..runtime_subscription.all_subscriptions.len()).rev() + { + let sender = + runtime_subscription.all_subscriptions.swap_remove(index); + let notif = super::Notification::Finalized { + hash, + best_block_hash_if_changed: if best_output_block_updated { + Some(best_block_hash) + } else { + None + }, + pruned_blocks: pruned_blocks_hashes.clone(), + }; + if sender.try_send(notif).is_ok() { + runtime_subscription.all_subscriptions.push(sender); + } + } + } + + async_tree::OutputUpdate::Finalized { .. } + | async_tree::OutputUpdate::BestBlockChanged { .. } => { + // Do not report anything to subscriptions if no finalized parahead is + // known yet. + let finalized_parahead = match runtime_subscription + .async_tree + .output_finalized_async_user_data() + { + Some(p) => p, + None => continue, + }; + + // Calculate hash of the parablock corresponding to the new best relay + // chain block. + let parahash = header::hash_from_scale_encoded_header( + runtime_subscription + .async_tree + .output_best_block_index() + .map(|(_, b)| b.as_ref().unwrap()) + .unwrap_or(finalized_parahead), + ); + + if runtime_subscription.reported_best_parahead_hash.as_ref() + != Some(¶hash) + { + runtime_subscription.reported_best_parahead_hash = + Some(parahash); + + log!( + &self.platform, + Debug, + &self.log_target, + "subscriptions-notify-best-block-changed", + hash = HashDisplay(¶hash) + ); + + // Elements in `all_subscriptions` are removed one by one and + // inserted back if the channel is still open. + for index in + (0..runtime_subscription.all_subscriptions.len()).rev() + { + let sender = runtime_subscription + .all_subscriptions + .swap_remove(index); + let notif = super::Notification::BestBlockChanged { + hash: parahash, + }; + if sender.try_send(notif).is_ok() { + runtime_subscription.all_subscriptions.push(sender); + } + } + } + } + + async_tree::OutputUpdate::Block(block) => { + // `block` borrows `async_tree`. We need to mutably access `async_tree` + // below, so deconstruct `block` beforehand. + let is_new_best = block.is_new_best; + let block_index = block.index; + let scale_encoded_header: Vec = runtime_subscription + .async_tree + .block_async_user_data(block.index) + .unwrap() + .clone() + .unwrap(); + let parahash = + header::hash_from_scale_encoded_header(&scale_encoded_header); + + // Do not report anything to subscriptions if no finalized parahead is + // known yet. + let finalized_parahead = match runtime_subscription + .async_tree + .output_finalized_async_user_data() + { + Some(p) => p, + None => continue, + }; + + // Do not report the new block if it has already been reported in the + // past. This covers situations where the parahead is identical to the + // relay chain's parent's parahead, but also situations where multiple + // sibling relay chain blocks have the same parahead. + if *finalized_parahead == scale_encoded_header + || runtime_subscription + .async_tree + .input_output_iter_unordered() + .filter(|item| item.id != block_index) + .filter_map(|item| item.async_op_user_data) + .any(|item| item.as_ref() == Some(&scale_encoded_header)) + { + // While the parablock has already been reported, it is possible that + // it becomes the new best block while it wasn't before, in which + // case we should send a notification. + if is_new_best + && runtime_subscription.reported_best_parahead_hash.as_ref() + != Some(¶hash) + { + runtime_subscription.reported_best_parahead_hash = + Some(parahash); + + log!( + &self.platform, + Debug, + &self.log_target, + "subscriptions-notify-best-block-changed", + hash = HashDisplay(¶hash) + ); + + // Elements in `all_subscriptions` are removed one by one and + // inserted back if the channel is still open. + for index in + (0..runtime_subscription.all_subscriptions.len()).rev() + { + let sender = runtime_subscription + .all_subscriptions + .swap_remove(index); + let notif = super::Notification::BestBlockChanged { + hash: parahash, + }; + if sender.try_send(notif).is_ok() { + runtime_subscription.all_subscriptions.push(sender); + } + } + } + + continue; + } + + if is_new_best { + runtime_subscription.reported_best_parahead_hash = + Some(parahash); + } + + let parent_hash = header::hash_from_scale_encoded_header( + runtime_subscription + .async_tree + .parent(block_index) + .map(|idx| { + runtime_subscription + .async_tree + .block_async_user_data(idx) + .unwrap() + .as_ref() + .unwrap() + }) + .unwrap_or(finalized_parahead), + ); + + log!( + &self.platform, + Debug, + &self.log_target, + "subscriptions-notify-new-parablock", + hash = HashDisplay(¶hash), + parent_hash = HashDisplay(&parent_hash), + ?is_new_best + ); + + // Elements in `all_subscriptions` are removed one by one and + // inserted back if the channel is still open. + for index in (0..runtime_subscription.all_subscriptions.len()).rev() + { + let sender = + runtime_subscription.all_subscriptions.swap_remove(index); + let notif = + super::Notification::Block(super::BlockNotification { + is_new_best, + parent_hash, + scale_encoded_header: scale_encoded_header.clone(), + }); + if sender.try_send(notif).is_ok() { + runtime_subscription.all_subscriptions.push(sender); + } + } + } + } + } + } + + ( + WakeUpReason::StartParaheadFetch, + ParachainBackgroundState::Subscribed(runtime_subscription), + ) => { + // Must start downloading a parahead. + + // Internal state check. + debug_assert_eq!( + runtime_subscription.reported_best_parahead_hash.is_some(), + runtime_subscription + .async_tree + .output_finalized_async_user_data() + .is_some() + ); + + // Limit the maximum number of simultaneous downloads. + if runtime_subscription.in_progress_paraheads.len() >= 4 { + continue; + } + + match runtime_subscription + .async_tree + .next_necessary_async_op(&self.platform.now()) + { + async_tree::NextNecessaryAsyncOp::NotReady { when: Some(when) } => { + runtime_subscription.next_start_parahead_fetch = + Box::pin(self.platform.sleep_until(when)); + } + async_tree::NextNecessaryAsyncOp::NotReady { when: None } => { + runtime_subscription.next_start_parahead_fetch = + Box::pin(future::pending()); + } + async_tree::NextNecessaryAsyncOp::Ready(op) => { + log!( + &self.platform, + Debug, + &self.log_target, + "parahead-fetch-operation-started", + relay_block_hash = + HashDisplay(&runtime_subscription.async_tree[op.block_index]), + ); + + runtime_subscription.in_progress_paraheads.push({ + let relay_chain_sync = self.relay_chain_sync.clone(); + let subscription_id = + runtime_subscription.relay_chain_subscribe_all.id(); + let block_hash = runtime_subscription.async_tree[op.block_index]; + let async_op_id = op.id; + let parachain_id = self.parachain_id; + Box::pin(async move { + ( + async_op_id, + fetch_parahead( + &relay_chain_sync, + subscription_id, + parachain_id, + &block_hash, + ) + .await, + ) + }) + }); + + // There might be more downloads to start. + runtime_subscription.next_start_parahead_fetch = + Box::pin(future::ready(())); + } + } + } + + ( + WakeUpReason::Notification(runtime_service::Notification::Finalized { + hash, + best_block_hash_if_changed, + .. + }), + ParachainBackgroundState::Subscribed(runtime_subscription), + ) => { + // Relay chain has a new finalized block. + log!( + &self.platform, + Debug, + &self.log_target, + "relay-chain-block-finalized", + hash = HashDisplay(&hash) + ); + + if let Some(best_block_hash_if_changed) = best_block_hash_if_changed { + let best = runtime_subscription + .async_tree + .input_output_iter_unordered() + .find(|b| *b.user_data == best_block_hash_if_changed) + .unwrap() + .id; + runtime_subscription + .async_tree + .input_set_best_block(Some(best)); + } + + let finalized = runtime_subscription + .async_tree + .input_output_iter_unordered() + .find(|b| *b.user_data == hash) + .unwrap() + .id; + runtime_subscription.async_tree.input_finalize(finalized); + runtime_subscription.must_process_sync_tree = true; + } + + ( + WakeUpReason::Notification(runtime_service::Notification::Block(block)), + ParachainBackgroundState::Subscribed(runtime_subscription), + ) => { + // Relay chain has a new block. + let hash = header::hash_from_scale_encoded_header(&block.scale_encoded_header); + + log!( + &self.platform, + Debug, + &self.log_target, + "relay-chain-new-block", + hash = HashDisplay(&hash), + parent_hash = HashDisplay(&block.parent_hash) + ); + + let parent = runtime_subscription + .async_tree + .input_output_iter_unordered() + .find(|b| *b.user_data == block.parent_hash) + .map(|b| b.id); // TODO: check if finalized + runtime_subscription.async_tree.input_insert_block( + hash, + parent, + false, + block.is_new_best, + ); + runtime_subscription.must_process_sync_tree = true; + + runtime_subscription.next_start_parahead_fetch = Box::pin(future::ready(())); + } + + ( + WakeUpReason::Notification(runtime_service::Notification::BestBlockChanged { + hash, + }), + ParachainBackgroundState::Subscribed(runtime_subscription), + ) => { + // Relay chain has a new best block. + log!( + &self.platform, + Debug, + &self.log_target, + "relay-chain-best-block-changed", + hash = HashDisplay(&hash) + ); + + // If the block isn't found in `async_tree`, assume that it is equal to the + // finalized block (that has left the tree already). + let node_idx = runtime_subscription + .async_tree + .input_output_iter_unordered() + .find(|b| *b.user_data == hash) + .map(|b| b.id); + runtime_subscription + .async_tree + .input_set_best_block(node_idx); + + runtime_subscription.must_process_sync_tree = true; + } + + (WakeUpReason::SubscriptionDead, _) => { + // Recreate the channel. + log!( + &self.platform, + Debug, + &self.log_target, + "relay-chain-subscription-reset" + ); + self.subscription_state = ParachainBackgroundState::NotSubscribed { + all_subscriptions: Vec::new(), + subscribe_future: { + let relay_chain_sync = self.relay_chain_sync.clone(); + Box::pin(async move { + relay_chain_sync + .subscribe_all(32, NonZero::::new(usize::MAX).unwrap()) + .await + }) + }, + }; + } + + ( + WakeUpReason::ParaheadFetchFinished { + async_op_id, + parahead_result: Ok(parahead), + }, + ParachainBackgroundState::Subscribed(runtime_subscription), + ) => { + // A parahead fetching operation is successful. + log!( + &self.platform, + Debug, + &self.log_target, + "parahead-fetch-operation-success", + parahead_hash = HashDisplay( + blake2_rfc::blake2b::blake2b(32, b"", ¶head).as_bytes() + ), + relay_blocks = runtime_subscription + .async_tree + .async_op_blocks(async_op_id) + .map(|b| HashDisplay(b)) + .join(",") + ); + + // Unpin the relay blocks whose parahead is now known. + for block in runtime_subscription + .async_tree + .async_op_finished(async_op_id, Some(parahead)) + { + let hash = &runtime_subscription.async_tree[block]; + runtime_subscription + .relay_chain_subscribe_all + .unpin_block(*hash) + .await; + } + + runtime_subscription.must_process_sync_tree = true; + + runtime_subscription.next_start_parahead_fetch = Box::pin(future::ready(())); + } + + ( + WakeUpReason::ParaheadFetchFinished { + parahead_result: + Err(ParaheadError::PinRuntimeError( + runtime_service::PinPinnedBlockRuntimeError::ObsoleteSubscription, + )), + .. + }, + _, + ) => { + // The relay chain runtime service has some kind of gap or issue and has + // discarded the runtime. + // Destroy the subscription and recreate the channels. + log!( + &self.platform, + Debug, + &self.log_target, + "relay-chain-subscription-reset" + ); + self.subscription_state = ParachainBackgroundState::NotSubscribed { + all_subscriptions: Vec::new(), + subscribe_future: { + let relay_chain_sync = self.relay_chain_sync.clone(); + Box::pin(async move { + relay_chain_sync + .subscribe_all(32, NonZero::::new(usize::MAX).unwrap()) + .await + }) + }, + }; + } + + ( + WakeUpReason::ParaheadFetchFinished { + async_op_id, + parahead_result: Err(error), + }, + ParachainBackgroundState::Subscribed(runtime_subscription), + ) => { + // Failed fetching a parahead. + + // Several relay chains initially didn't support parachains, and have later + // been upgraded to support them. Similarly, the parachain might not have had a + // core on the relay chain until recently. For these reasons, errors when the + // relay chain is not near head of the chain are most likely normal and do + // not warrant logging an error. + // Note that `is_near_head_of_chain_heuristic` is normally not acceptable to + // use due to being too vague, but since this is just about whether to print a + // log message, it's completely fine. + if self + .relay_chain_sync + .is_near_head_of_chain_heuristic() + .await + && !error.is_network_problem() + { + log!( + &self.platform, + Error, + &self.log_target, + format!( + "Failed to fetch the parachain head from relay chain blocks {}: {}", + runtime_subscription + .async_tree + .async_op_blocks(async_op_id) + .map(|b| HashDisplay(b)) + .join(", "), + error + ) + ); + } + + log!( + &self.platform, + Debug, + &self.log_target, + "parahead-fetch-operation-error", + relay_blocks = runtime_subscription + .async_tree + .async_op_blocks(async_op_id) + .map(|b| HashDisplay(b)) + .join(","), + ?error + ); + + runtime_subscription + .async_tree + .async_op_failure(async_op_id, &self.platform.now()); + + runtime_subscription.next_start_parahead_fetch = Box::pin(future::ready(())); + } + + ( + WakeUpReason::ForegroundMessage(ToBackground::IsNearHeadOfChainHeuristic { + send_back, + }), + ParachainBackgroundState::Subscribed(sub), + ) if sub.async_tree.output_finalized_async_user_data().is_some() => { + // Since there is a mapping between relay chain blocks and parachain blocks, + // whether a parachain is at the head of the chain is the same thing as whether + // its relay chain is at the head of the chain. + // Note that there is no ordering guarantee of any kind w.r.t. block + // subscriptions notifications. + let val = self + .relay_chain_sync + .is_near_head_of_chain_heuristic() + .await; + let _ = send_back.send(val); + } + + ( + WakeUpReason::ForegroundMessage(ToBackground::IsNearHeadOfChainHeuristic { + send_back, + }), + _, + ) => { + // If no finalized parahead is known yet, we might be very close to the head + // but also maybe very very far away. We lean on the cautious side and always + // return `false`. + let _ = send_back.send(false); + } + + ( + WakeUpReason::ForegroundMessage(ToBackground::SubscribeAll { + send_back, + buffer_size, + .. + }), + ParachainBackgroundState::NotSubscribed { + all_subscriptions, .. + }, + ) => { + let (tx, new_blocks) = async_channel::bounded(buffer_size.saturating_sub(1)); + + // No known finalized parahead. + let _ = send_back.send(super::SubscribeAll { + finalized_block_scale_encoded_header: self + .obsolete_finalized_parahead + .clone(), + finalized_block_runtime: None, + non_finalized_blocks_ancestry_order: Vec::new(), + new_blocks, + }); + + all_subscriptions.push(tx); + } + + ( + WakeUpReason::ForegroundMessage(ToBackground::SubscribeAll { + send_back, + buffer_size, + .. + }), + ParachainBackgroundState::Subscribed(runtime_subscription), + ) => { + let (tx, new_blocks) = async_channel::bounded(buffer_size.saturating_sub(1)); + + // There are two possibilities here: either we know of any recent finalized + // parahead, or we don't. In case where we don't know of any finalized parahead + // yet, we report a single obsolete finalized parahead, which is + // `obsolete_finalized_parahead`. The rest of this module makes sure that no + // other block is reported to subscriptions as long as this is the case, and + // that subscriptions are reset once the first known finalized parahead + // is known. + if let Some(finalized_parahead) = runtime_subscription + .async_tree + .output_finalized_async_user_data() + { + // Finalized parahead is known. + let finalized_parahash = + header::hash_from_scale_encoded_header(finalized_parahead); + let _ = send_back.send(super::SubscribeAll { + finalized_block_scale_encoded_header: finalized_parahead.clone(), + finalized_block_runtime: None, + non_finalized_blocks_ancestry_order: { + let mut list = + Vec::<([u8; 32], super::BlockNotification)>::with_capacity( + runtime_subscription + .async_tree + .num_input_non_finalized_blocks(), + ); + + for relay_block in runtime_subscription + .async_tree + .input_output_iter_ancestry_order() + { + let parablock = match relay_block.async_op_user_data { + Some(b) => b.as_ref().unwrap(), + None => continue, + }; + + let parablock_hash = + header::hash_from_scale_encoded_header(parablock); + + // TODO: O(n) + if let Some((_, entry)) = + list.iter_mut().find(|(h, _)| *h == parablock_hash) + { + // Block is already in the list. Don't add it a second time. + if relay_block.is_output_best { + entry.is_new_best = true; + } + continue; + } + + // Find the parent of the parablock. This is done by going through + // the ancestors of the corresponding relay chain block (until and + // including the finalized relay chain block) until we find one + // whose parablock is different from the parablock in question. + // If none is found, the parablock is the same as the finalized + // parablock. + let parent_hash = runtime_subscription + .async_tree + .ancestors(relay_block.id) + .find_map(|idx| { + let hash = header::hash_from_scale_encoded_header( + runtime_subscription + .async_tree + .block_async_user_data(idx) + .unwrap() + .as_ref() + .unwrap(), + ); + if hash != parablock_hash { + Some(hash) + } else { + None + } + }) + .or_else(|| { + if finalized_parahash != parablock_hash { + Some(finalized_parahash) + } else { + None + } + }); + + // `parent_hash` is `None` if the parablock is + // the same as the finalized parablock, in which case we + // don't add it to the list. + if let Some(parent_hash) = parent_hash { + debug_assert!( + list.iter().filter(|(h, _)| *h == parent_hash).count() + == 1 + || parent_hash == finalized_parahash + ); + list.push(( + parablock_hash, + super::BlockNotification { + is_new_best: relay_block.is_output_best, + scale_encoded_header: parablock.clone(), + parent_hash, + }, + )); + } + } + + list.into_iter().map(|(_, v)| v).collect() + }, + new_blocks, + }); + } else { + // No known finalized parahead. + let _ = send_back.send(super::SubscribeAll { + finalized_block_scale_encoded_header: self + .obsolete_finalized_parahead + .clone(), + finalized_block_runtime: None, + non_finalized_blocks_ancestry_order: Vec::new(), + new_blocks, + }); + } + + runtime_subscription.all_subscriptions.push(tx); + } + + ( + WakeUpReason::ForegroundMessage(ToBackground::PeersAssumedKnowBlock { + send_back, + .. + }), + _, + ) => { + let _ = send_back.send(Vec::new()); + } + + (WakeUpReason::ForegroundMessage(ToBackground::SyncingPeers { send_back }), _) => { + let _ = send_back.send(Vec::new()); + } + + ( + WakeUpReason::ForegroundMessage(ToBackground::SerializeChainInformation { + send_back, + }), + _, + ) => { + let _ = send_back.send(None); + } + + ( + WakeUpReason::ParaheadFetchFinished { .. } + | WakeUpReason::AdvanceSyncTree + | WakeUpReason::Notification(_) + | WakeUpReason::StartParaheadFetch, + ParachainBackgroundState::NotSubscribed { .. }, + ) => { + // These paths are unreachable. + debug_assert!(false); + } + } + } + } +} + +async fn fetch_parahead( + relay_chain_sync: &Arc>, + subscription_id: runtime_service::SubscriptionId, + parachain_id: u32, + block_hash: &[u8; 32], +) -> Result, ParaheadError> { + // Call `ParachainHost_persisted_validation_data` in order to know where the parachain is. + let (pinned_runtime, block_state_trie_root, block_number) = relay_chain_sync + .pin_pinned_block_runtime(subscription_id, *block_hash) + .await + .map_err(ParaheadError::PinRuntimeError)?; + let success = relay_chain_sync + .runtime_call( + pinned_runtime, + *block_hash, + block_number, + block_state_trie_root, + para::PERSISTED_VALIDATION_FUNCTION_NAME.to_owned(), + None, // TODO: /!\ + para::persisted_validation_data_parameters( + parachain_id, + para::OccupiedCoreAssumption::TimedOut, + ) + .fold(Vec::new(), |mut a, b| { + a.extend_from_slice(b.as_ref()); + a + }), + 6, + Duration::from_secs(10), + NonZero::::new(2).unwrap(), + ) + .await + .map_err(ParaheadError::RuntimeCall)?; + + // Try decode the result of the runtime call. + // If this fails, it indicates an incompatibility between smoldot and the relay chain. + match para::decode_persisted_validation_data_return_value( + &success.output, + relay_chain_sync.block_number_bytes(), + ) { + Ok(Some(pvd)) => Ok(pvd.parent_head.to_vec()), + Ok(None) => Err(ParaheadError::NoCore), + Err(error) => Err(ParaheadError::InvalidRuntimeOutput(error)), + } +} + +/// Error that can happen when fetching the parachain head corresponding to a relay chain block. +#[derive(Debug, derive_more::Display, derive_more::Error)] +enum ParaheadError { + /// Error while performing call request over the network. + #[display("Error while performing call request over the network: {_0}")] + RuntimeCall(runtime_service::RuntimeCallError), + /// Error pinning the runtime of the block. + PinRuntimeError(runtime_service::PinPinnedBlockRuntimeError), + /// Parachain doesn't have a core in the relay chain. + NoCore, + /// Error while decoding the output of the call. + /// + /// This indicates some kind of incompatibility between smoldot and the relay chain. + #[display("Error while decoding the output of the call: {_0}")] + InvalidRuntimeOutput(para::Error), +} + +impl ParaheadError { + /// Returns `true` if this is caused by networking issues, as opposed to a consensus-related + /// issue. + fn is_network_problem(&self) -> bool { + match self { + ParaheadError::RuntimeCall(runtime_service::RuntimeCallError::Inaccessible(_)) => true, + ParaheadError::RuntimeCall( + runtime_service::RuntimeCallError::Execution(_) + | runtime_service::RuntimeCallError::Crash + | runtime_service::RuntimeCallError::InvalidRuntime(_) + | runtime_service::RuntimeCallError::ApiVersionRequirementUnfulfilled, + ) => false, + ParaheadError::PinRuntimeError(_) => false, + ParaheadError::NoCore => false, + ParaheadError::InvalidRuntimeOutput(_) => false, + } + } +}