pezkuwichain/pezkuwi-subxt
1
0
Fork 0
mirror of https://github.com/pezkuwichain/pezkuwi-subxt.git synced 2026-06-11 08:21:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

WIP: Combined backend done, client in progress

Browse Source
This commit is contained in:
James Wilson
2025-12-01 16:18:45 +00:00
parent b216240575
commit a3381207d2
14 changed files with 439 additions and 222 deletions
Download Patch File Download Diff File Expand all files Collapse all files
new/src/backend.rs
+26 -1
View File
@@ -21,6 +21,25 @@ use std::pin::Pin;
use std::sync::Arc;
use subxt_metadata::Metadata;
// Expose our various backends.
pub use chain_head::{
ChainHeadBackend,
ChainHeadBackendBuilder,
ChainHeadBackendDriver,
};
pub use archive::{
ArchiveBackend
};
pub use legacy::{
LegacyBackend,
LegacyBackendBuilder
};
pub use combined::{
CombinedBackend,
CombinedBackendBuilder,
CombinedBackendDriver
};
/// Prevent the backend trait being implemented externally.
#[doc(hidden)]
pub(crate) mod sealed {
@@ -55,7 +74,13 @@ pub trait Backend<T: Config>: sealed::Sealed + Send + Sync + 'static {
/// Fetch the genesis hash
async fn genesis_hash(&self) -> Result<HashFor<T>, BackendError>;
/// Get a block header
/// Convert a block number to a hash. This should return `None` in the event that
/// multiple block hashes correspond to the given number (ie if the number is greater
/// than that of the latest finalized block and some forks exist). Nevertheless, it could
/// still return the hash to a block on some fork that is pruned.
async fn block_number_to_hash(&self, number: u64) -> Result<Option<BlockRef<HashFor<T>>>, BackendError>;
/// Get a block header.
async fn block_header(&self, at: HashFor<T>) -> Result<Option<T::Header>, BackendError>;
/// Return the extrinsics found in the block. Each extrinsic is represented
new/src/backend/archive.rs
+16 -5
View File
@@ -12,6 +12,7 @@
mod storage_stream;
use subxt_rpcs::methods::ChainHeadRpcMethods;
use crate::backend::{
Backend, BlockRef, StorageResponse, StreamOf, StreamOfResults,
TransactionStatus, utils::retry,
@@ -26,20 +27,17 @@ use subxt_rpcs::methods::chain_head::{
};
use storage_stream::ArchiveStorageStream;
// Expose the RPC methods.
pub use subxt_rpcs::methods::chain_head::ChainHeadRpcMethods as ArchiveRpcMethods;
/// The archive backend.
#[derive(Debug, Clone)]
pub struct ArchiveBackend<T: Config> {
// RPC methods we'll want to call:
methods: ArchiveRpcMethods<RpcConfigFor<T>>,
methods: ChainHeadRpcMethods<RpcConfigFor<T>>,
}
impl<T: Config> ArchiveBackend<T> {
/// Configure and construct an [`ArchiveBackend`] and the associated [`ChainHeadBackendDriver`].
pub fn new(client: impl Into<RpcClient>,) -> ArchiveBackend<T> {
let methods = ArchiveRpcMethods::new(client.into());
let methods = ChainHeadRpcMethods::new(client.into());
ArchiveBackend { methods }
}
@@ -123,6 +121,19 @@ impl<T: Config> Backend<T> for ArchiveBackend<T> {
.await
}
async fn block_number_to_hash(&self, number: u64) -> Result<Option<BlockRef<HashFor<T>>>, BackendError> {
retry(|| async {
let mut hashes = self.methods.archive_v1_hash_by_height(number as usize).await?;
if let (Some(hash), None) = (hashes.pop(), hashes.pop()) {
// One hash; return it.
Ok(Some(BlockRef::from_hash(hash)))
} else {
// More than one; return None.
Ok(None)
}
}).await
}
async fn block_header(&self, at: HashFor<T>) -> Result<Option<T::Header>, BackendError> {
retry(|| async {
let header = self.methods.archive_v1_header(at).await?;
new/src/backend/archive/storage_stream.rs
+3 -3
View File
@@ -7,11 +7,11 @@ use futures::{FutureExt, Stream, StreamExt};
use std::task::{Context, Poll};
use crate::error::BackendError;
use crate::config::{Config, HashFor, RpcConfigFor};
use super::ArchiveRpcMethods;
use subxt_rpcs::methods::ChainHeadRpcMethods;
pub struct ArchiveStorageStream<T: Config> {
at: HashFor<T>,
methods: ArchiveRpcMethods<RpcConfigFor<T>>,
methods: ChainHeadRpcMethods<RpcConfigFor<T>>,
query_queue: VecDeque<ArchiveStorageQuery<Vec<u8>>>,
state: Option<StreamState<T>>,
}
@@ -31,7 +31,7 @@ impl <T: Config> ArchiveStorageStream<T> {
/// Fetch descendant keys.
pub fn new(
at: HashFor<T>,
methods: ArchiveRpcMethods<RpcConfigFor<T>>,
methods: ChainHeadRpcMethods<RpcConfigFor<T>>,
query_queue: VecDeque<ArchiveStorageQuery<Vec<u8>>>,
) -> Self {
Self {
new/src/backend/chain_head.rs
+5 -3
View File
@@ -13,6 +13,7 @@ mod follow_stream_driver;
mod follow_stream_unpin;
mod storage_items;
use subxt_rpcs::methods::ChainHeadRpcMethods;
use self::follow_stream_driver::FollowStreamFinalizedHeads;
use crate::backend::{
Backend, BlockRef, BlockRefT, StorageResponse, StreamOf, StreamOfResults,
@@ -32,9 +33,6 @@ use subxt_rpcs::methods::chain_head::{
FollowEvent, MethodResponse, StorageQuery, StorageQueryType, StorageResultType,
};
// Expose the RPC methods.
pub use subxt_rpcs::methods::chain_head::ChainHeadRpcMethods;
/// Configure and build an [`ChainHeadBackend`].
pub struct ChainHeadBackendBuilder<T> {
max_block_life: usize,
@@ -375,6 +373,10 @@ impl<T: Config> Backend<T> for ChainHeadBackend<T> {
.await
}
async fn block_number_to_hash(&self, _number: u64) -> Result<Option<BlockRef<HashFor<T>>>, BackendError> {
Err(BackendError::other("The ChainHead V1 RPCs do not support obtaining a block hash from a number."))
}
async fn block_header(&self, at: HashFor<T>) -> Result<Option<T::Header>, BackendError> {
retry(|| async {
let sub_id = get_subscription_id(&self.follow_handle).await?;
new/src/backend/combined.rs
+171 -45
View File
@@ -8,19 +8,16 @@
use crate::backend::chain_head::ChainHeadBackendDriver;
use crate::backend::{
legacy::LegacyBackend,
chain_head::{ChainHeadBackend, },
chain_head::ChainHeadBackend,
archive::ArchiveBackend,
Backend, BlockRef, StorageResponse, StreamOf, StreamOfResults,
TransactionStatus, utils::retry,
Backend, BlockRef, StorageResponse, StreamOfResults,
TransactionStatus,
};
use crate::config::{Config, HashFor, RpcConfigFor};
use crate::error::BackendError;
use crate::config::{Config, HashFor};
use crate::error::{BackendError, CombinedBackendError};
use async_trait::async_trait;
use futures::StreamExt;
use subxt_rpcs::RpcClient;
use subxt_rpcs::methods::chain_head::{
ArchiveStorageQuery, ArchiveCallResult, StorageQueryType,
};
use futures::Stream;
use std::task::Poll;
@@ -89,13 +86,14 @@ impl <T: Config> CombinedBackendBuilder<T> {
///
/// If you just want to run the driver in the background until completion in on the default runtime,
/// use [`CombinedBackendBuilder::build_with_background_driver`] instead.
pub async fn build(self, rpc_client: impl Into<RpcClient>) -> Result<(CombinedBackend<T>, CombinedBackendDriver<T>), BackendError> {
pub async fn build(self, rpc_client: impl Into<RpcClient>) -> Result<(CombinedBackend<T>, CombinedBackendDriver<T>), CombinedBackendError> {
let rpc_client = rpc_client.into();
// What does the thing wer're talking to actually know about?
let methods: Vec<String> = rpc_client
.request("rpc_methods", subxt_rpcs::rpc_params![])
.await?;
.await
.map_err(CombinedBackendError::CouldNotObtainRpcMethodList)?;
let has_archive_methods = methods.iter().any(|m| m.starts_with("archive_v1_"));
let has_chainhead_methods = methods.iter().any(|m| m.starts_with("chainHead_v1"));
@@ -143,8 +141,10 @@ impl <T: Config> CombinedBackendBuilder<T> {
/// - On non-wasm targets, this will spawn the driver on `tokio`.
/// - On wasm targets, this will spawn the driver on `wasm-bindgen-futures`.
#[cfg(feature = "runtime")]
pub async fn build_with_background_driver(self, client: impl Into<RpcClient>) -> Result<CombinedBackend<T>, BackendError> {
let (backend, mut driver) = self.build(client).await?;
pub async fn build_with_background_driver(self, client: impl Into<RpcClient>) -> Result<CombinedBackend<T>, CombinedBackendError> {
let (backend, mut driver) = self
.build(client)
.await?;
super::utils::spawn(async move {
// NOTE: we need to poll the driver until it's done i.e returns None
@@ -201,29 +201,41 @@ impl <T: Config> CombinedBackend<T> {
pub fn builder() -> CombinedBackendBuilder<T> {
CombinedBackendBuilder::new()
}
fn archive(&self) -> Option<&dyn Backend<T>> {
self.archive.as_ref().map(|a| {
let a: &dyn Backend<T> = a;
a
})
}
fn chainhead(&self) -> Option<&dyn Backend<T>> {
self.chainhead.as_ref().map(|a| {
let a: &dyn Backend<T> = a;
a
})
}
fn legacy(&self) -> Option<&dyn Backend<T>> {
self.legacy.as_ref().map(|a| {
let a: &dyn Backend<T> = a;
a
})
}
}
impl<T: Config> super::sealed::Sealed for CombinedBackend<T> {}
static NO_AVAILABLE_BACKEND: &str = "No available RPC methods to use. `no_default_backends` was used, but no applicable backends were then provided.";
macro_rules! call_backends {
({$($backend_name:ident)|+}. $method_name:ident ( $($arg:expr),* )) => {{
let mut err = BackendError::other(NO_AVAILABLE_BACKEND);
$(
if let Some(backend) = &self.$backend_name {
err = match backend.$method_name($( $arg, )*).await {
Ok(res) => return Ok(res),
Err(e) => e
}
}
)+
return Err(err)
}}
}
// Our default behaviour:
// - Try the archive backend first if it's available. Why? It has all block headers/bodies
// etc so it's mroe likely to succeed than chainHead backend and give back things that won't
// expire.
// - If archive calls aren't available, fall back to the chainHead backend. Blocks given back
// by this are more likely to expire.
// - If neither exists / works, we fall back to the legacy methods. These have some limits on
// what is available (often fewer limits than chainHead though) but tend to do the job. We'd
// rather not use these as they are old and should go away, but until then they are a good
// fallback.
#[async_trait]
impl<T: Config> Backend<T> for CombinedBackend<T> {
async fn storage_fetch_values(
@@ -231,7 +243,13 @@ impl<T: Config> Backend<T> for CombinedBackend<T> {
keys: Vec<Vec<u8>>,
at: HashFor<T>,
) -> Result<StreamOfResults<StorageResponse>, BackendError> {
call_backends!({archive|chainhead|legacy}.storage_fetch_values(keys, at))
try_backends(&[
self.archive(),
self.chainhead(),
self.legacy()
], async |b: &dyn Backend<T>| {
b.storage_fetch_values(keys.clone(), at).await
}).await
}
async fn storage_fetch_descendant_keys(
@@ -239,7 +257,13 @@ impl<T: Config> Backend<T> for CombinedBackend<T> {
key: Vec<u8>,
at: HashFor<T>,
) -> Result<StreamOfResults<Vec<u8>>, BackendError> {
try_backends(&[
self.archive(),
self.chainhead(),
self.legacy()
], async |b: &dyn Backend<T>| {
b.storage_fetch_descendant_keys(key.clone(), at).await
}).await
}
async fn storage_fetch_descendant_values(
@@ -247,51 +271,123 @@ impl<T: Config> Backend<T> for CombinedBackend<T> {
key: Vec<u8>,
at: HashFor<T>,
) -> Result<StreamOfResults<StorageResponse>, BackendError> {
try_backends(&[
self.archive(),
self.chainhead(),
self.legacy()
], async |b: &dyn Backend<T>| {
b.storage_fetch_descendant_values(key.clone(), at).await
}).await
}
async fn genesis_hash(&self) -> Result<HashFor<T>, BackendError> {
try_backends(&[
self.archive(),
self.chainhead(),
self.legacy()
], async |b: &dyn Backend<T>| {
b.genesis_hash().await
}).await
}
async fn block_number_to_hash(&self, number: u64) -> Result<Option<BlockRef<HashFor<T>>>, BackendError> {
try_backends(&[
self.archive(),
self.legacy(),
// chainHead last as it cannot handle this request and will fail, so it's here
// just to hand back a more relevant error in case the above two backends aren't
// enabled or have some issue.
self.chainhead()
], async |b: &dyn Backend<T>| {
b.block_number_to_hash(number).await
}).await
}
async fn block_header(&self, at: HashFor<T>) -> Result<Option<T::Header>, BackendError> {
try_backends(&[
self.archive(),
self.chainhead(),
self.legacy()
], async |b: &dyn Backend<T>| {
b.block_header(at).await
}).await
}
async fn block_body(&self, at: HashFor<T>) -> Result<Option<Vec<Vec<u8>>>, BackendError> {
try_backends(&[
self.archive(),
self.chainhead(),
self.legacy()
], async |b: &dyn Backend<T>| {
b.block_body(at).await
}).await
}
async fn latest_finalized_block_ref(&self) -> Result<BlockRef<HashFor<T>>, BackendError> {
try_backends(&[
// Prioritize chainHead backend since it's streaming these things; save another call.
self.chainhead(),
self.archive(),
self.legacy()
], async |b: &dyn Backend<T>| {
b.latest_finalized_block_ref().await
}).await
}
async fn stream_all_block_headers(
&self,
_hasher: T::Hasher,
hasher: T::Hasher,
) -> Result<StreamOfResults<(T::Header, BlockRef<HashFor<T>>)>, BackendError> {
try_backends(&[
// Ignore archive backend; it doesn't support this.
self.chainhead(),
self.legacy()
], async |b: &dyn Backend<T>| {
b.stream_all_block_headers(hasher.clone()).await
}).await
}
async fn stream_best_block_headers(
&self,
_hasher: T::Hasher,
hasher: T::Hasher,
) -> Result<StreamOfResults<(T::Header, BlockRef<HashFor<T>>)>, BackendError> {
try_backends(&[
// Ignore archive backend; it doesn't support this.
self.chainhead(),
self.legacy()
], async |b: &dyn Backend<T>| {
b.stream_best_block_headers(hasher.clone()).await
}).await
}
async fn stream_finalized_block_headers(
&self,
_hasher: T::Hasher,
hasher: T::Hasher,
) -> Result<StreamOfResults<(T::Header, BlockRef<HashFor<T>>)>, BackendError> {
try_backends(&[
// Ignore archive backend; it doesn't support this.
self.chainhead(),
self.legacy()
], async |b: &dyn Backend<T>| {
b.stream_finalized_block_headers(hasher.clone()).await
}).await
}
async fn submit_transaction(
&self,
extrinsic: &[u8],
) -> Result<StreamOfResults<TransactionStatus<HashFor<T>>>, BackendError> {
try_backends(&[
// chainHead first as it does the same as the archive backend, but with better
// guarantees around the block handed back being pinned & ready to access.
self.chainhead(),
self.legacy(),
// archive last just incase chainHead & legacy fail or aren't provided for some
// reason.
self.archive(),
], async |b: &dyn Backend<T>| {
b.submit_transaction(extrinsic).await
}).await
}
async fn call(
@@ -300,6 +396,36 @@ impl<T: Config> Backend<T> for CombinedBackend<T> {
call_parameters: Option<&[u8]>,
at: HashFor<T>,
) -> Result<Vec<u8>, BackendError> {
try_backends(&[
self.archive(),
self.chainhead(),
self.legacy()
], async |b: &dyn Backend<T>| {
b.call(method, call_parameters, at).await
}).await
}
}
/// Call one backend after the other in the list until we get a successful result back.
async fn try_backends<'s, 'b, T, Func, Fut, O>(
backends: &'s [Option<&'b dyn Backend<T>>],
mut f: Func
) -> Result<O, BackendError>
where
'b: 's,
T: Config,
Func: FnMut(&'b dyn Backend<T>) -> Fut,
Fut: Future<Output = Result<O, BackendError>> + 'b,
{
static NO_AVAILABLE_BACKEND: &str = "None of the configured backends are capable of handling this request";
let mut err = BackendError::other(NO_AVAILABLE_BACKEND);
for backend in backends.into_iter().filter_map(|b| *b) {
match f(backend).await {
Ok(res) => return Ok(res),
Err(e) => { err = e }
}
}
Err(err)
}
new/src/backend/legacy.rs
+14 -9
View File
@@ -7,7 +7,7 @@
mod descendant_streams;
use self::rpc_methods::TransactionStatus as RpcTransactionStatus;
use subxt_rpcs::methods::legacy::{ TransactionStatus as RpcTransactionStatus, LegacyRpcMethods };
use crate::backend::utils::{retry, retry_stream};
use crate::backend::{
Backend, BlockRef, StorageResponse, StreamOf, StreamOfResults,
@@ -19,17 +19,10 @@ use async_trait::async_trait;
use futures::TryStreamExt;
use futures::{Future, Stream, StreamExt, future, future::Either, stream};
use subxt_rpcs::RpcClient;
use subxt_rpcs::methods::legacy::NumberOrHex;
use codec::Encode;
use descendant_streams::{StorageFetchDescendantKeysStream, StorageFetchDescendantValuesStream};
/// Re-export legacy RPC types and methods from [`subxt_rpcs::methods::legacy`].
pub mod rpc_methods {
pub use subxt_rpcs::methods::legacy::*;
}
// Expose the RPC methods.
pub use rpc_methods::LegacyRpcMethods;
/// Configure and build an [`LegacyBackend`].
pub struct LegacyBackendBuilder<T> {
storage_page_size: u32,
@@ -184,6 +177,18 @@ impl<T: Config> Backend<T> for LegacyBackend<T> {
.await
}
async fn block_number_to_hash(&self, number: u64) -> Result<Option<BlockRef<HashFor<T>>>, BackendError> {
retry(|| async {
let number_or_hash = NumberOrHex::Number(number);
let hash = self
.methods
.chain_get_block_hash(Some(number_or_hash))
.await?
.map(BlockRef::from_hash);
Ok(hash)
}).await
}
async fn block_header(&self, at: HashFor<T>) -> Result<Option<T::Header>, BackendError> {
retry(|| async {
let header = self.methods.chain_get_header(Some(at)).await?;
new/src/backend/utils.rs
+1 -7
View File
@@ -118,13 +118,6 @@ where
})))
}
/// Resubscribe callback.
type ResubscribeGetter<T> = Box<dyn FnMut() -> ResubscribeFuture<T> + Send>;
/// Future that resolves to a subscription stream.
type ResubscribeFuture<T> =
Pin<Box<dyn Future<Output = Result<StreamOfResults<T>, BackendError>> + Send>>;
/// Retry subscription.
struct RetrySubscription<F, R, T> {
resubscribe: F,
@@ -164,6 +157,7 @@ where
return Poll::Ready(Some(Err(err)));
}
Poll::Ready(None) => {
self.state = RetrySubscriptionState::Done;
return Poll::Ready(None)
}
Poll::Ready(Some(Ok(val))) => {
new/src/client/offline_client.rs
+2 -1
View File
@@ -22,8 +22,9 @@ impl<T: Config> OfflineClient<T> {
/// [`OfflineClient`] it's called on, and so cannot outlive it.
pub fn at(
&self,
block_number: u32,
block_number: impl Into<u64>,
) -> Result<ClientAtBlock<OfflineClientAtBlock, T>, OfflineClientAtBlockError> {
let block_number = block_number.into();
let spec_version = self
.config
.spec_version_for_block_number(block_number)
new/src/client/online_client.rs
+102 -119
View File
@@ -1,11 +1,13 @@
mod block_number_or_ref;
use super::ClientAtBlock;
use super::OfflineClientAtBlockT;
use crate::config::Header;
use crate::config::{ Config, HashFor, RpcConfigFor };
use crate::error::OnlineClientAtBlockError;
use crate::backend::Backend;
use crate::backend::{ Backend, CombinedBackend, BlockRef };
use codec::{Compact, Decode, Encode};
use frame_metadata::{RuntimeMetadata, RuntimeMetadataPrefixed};
use scale_info_legacy::TypeRegistrySet;
use std::sync::Arc;
use subxt_rpcs::methods::chain_head::ArchiveCallResult;
use subxt_rpcs::{ChainHeadRpcMethods, RpcClient};
@@ -15,6 +17,8 @@ use subxt_metadata::Metadata;
#[cfg_attr(docsrs, doc(cfg(feature = "jsonrpsee")))]
use crate::error::OnlineClientError;
pub use block_number_or_ref::BlockNumberOrRef;
/// A client which exposes the means to decode historic data on a chain online.
#[derive(Clone, Debug)]
pub struct OnlineClient<T: Config> {
@@ -42,14 +46,14 @@ impl<T: Config> OnlineClient<T> {
/// point to a locally running node on `ws://127.0.0.1:9944`.
///
/// **Note:** This will only work if the local node is an archive node.
#[cfg(feature = "jsonrpsee")]
#[cfg(all(feature = "jsonrpsee", feature = "runtime"))]
pub async fn new(config: T) -> Result<OnlineClient<T>, OnlineClientError> {
let url = "ws://127.0.0.1:9944";
OnlineClient::from_url(config, url).await
}
/// Construct a new [`OnlineClient`], providing a URL to connect to.
#[cfg(feature = "jsonrpsee")]
#[cfg(all(feature = "jsonrpsee", feature = "runtime"))]
pub async fn from_url(
config: T,
url: impl AsRef<str>,
@@ -69,13 +73,13 @@ impl<T: Config> OnlineClient<T> {
/// Construct a new [`OnlineClient`], providing a URL to connect to.
///
/// Allows insecure URLs without SSL encryption, e.g. (http:// and ws:// URLs).
#[cfg(feature = "jsonrpsee")]
#[cfg(all(feature = "jsonrpsee", feature = "runtime"))]
pub async fn from_insecure_url(
config: T,
url: impl AsRef<str>,
) -> Result<OnlineClient<T>, OnlineClientError> {
let rpc_client = RpcClient::from_insecure_url(url).await?;
Ok(OnlineClient::from_rpc_client(config, rpc_client))
OnlineClient::from_rpc_client(config, rpc_client).await
}
fn is_url_secure(url: &url::Url) -> bool {
@@ -90,19 +94,23 @@ impl<T: Config> OnlineClient<T> {
/// Construct a new [`OnlineClient`] by providing an [`RpcClient`] to drive the connection.
/// This will use the current default [`Backend`], which may change in future releases.
#[cfg(feature = "jsonrpsee")]
pub fn from_rpc_client(
#[cfg(all(feature = "jsonrpsee", feature = "runtime"))]
pub async fn from_rpc_client(
config: T,
rpc_client: impl Into<RpcClient>,
) -> OnlineClient<T> {
) -> Result<OnlineClient<T>, OnlineClientError> {
let rpc_client = rpc_client.into();
let backend = Arc::new(LegacyBackend::builder().build(rpc_client));
OnlineClient::from_backend(config, backend)
let backend = CombinedBackend::builder()
.build_with_background_driver(rpc_client)
.await
.map_err(OnlineClientError::CannotBuildCombinedBackend)?;
let backend: Arc<dyn Backend<T>> = Arc::new(backend);
Ok(OnlineClient::from_backend(config, backend))
}
/// Construct a new [`OnlineClient`] by providing an underlying [`Backend`]
/// implementation to power it.
pub fn from_backend<B: Backend<T>>(
pub fn from_backend(
config: T,
backend: impl Into<Arc<dyn Backend<T>>>,
) -> OnlineClient<T> {
@@ -118,56 +126,86 @@ impl<T: Config> OnlineClient<T> {
/// [`OnlineClient`] it's called on, and so cannot outlive it.
pub async fn at_block(
&self,
block_number: u32,
number_or_hash: impl Into<BlockNumberOrRef<T>>,
) -> Result<ClientAtBlock<OnlineClientAtBlock<T>, T>, OnlineClientAtBlockError> {
let config = &self.inner.config;
let rpc_methods = &self.inner.rpc_methods;
let number_or_hash = number_or_hash.into();
let block_hash = rpc_methods
.archive_v1_hash_by_height(block_number as usize)
.await
.map_err(|e| OnlineClientAtBlockError::CannotGetBlockHash {
block_number,
reason: e,
})?
.pop()
.ok_or_else(|| OnlineClientAtBlockError::BlockNotFound { block_number })?
.into();
// Get our configuration, or fetch from the node if not available.
let spec_version =
if let Some(spec_version) = config.spec_version_for_block_number(block_number) {
spec_version
} else {
// Fetch spec version. Caching this doesn't really make sense, so either
// details are provided offline or we fetch them every time.
get_spec_version(rpc_methods, block_hash).await?
};
let metadata = if let Some(metadata) = config.metadata_for_spec_version(spec_version) {
metadata
} else {
// Fetch and then give our config the opportunity to cache this metadata.
let metadata = get_metadata(rpc_methods, block_hash).await?;
let metadata = Arc::new(metadata);
config.set_metadata_for_spec_version(spec_version, metadata.clone());
metadata
// We are given either a block hash or number. We need both.
let (block_ref, block_num) = match number_or_hash {
BlockNumberOrRef::BlockRef(block_ref) => {
let block_hash = block_ref.hash();
let block_header = self
.inner
.backend
.block_header(block_hash)
.await
.map_err(|e| OnlineClientAtBlockError::CannotGetBlockHeader {
block_hash: block_hash.into(),
reason: e
})?
.ok_or(OnlineClientAtBlockError::BlockHeaderNotFound {
block_hash: block_hash.into()
})?;
(block_ref, block_header.number())
},
BlockNumberOrRef::Number(block_num) => {
let block_ref = self
.inner
.backend
.block_number_to_hash(block_num)
.await
.map_err(|e| OnlineClientAtBlockError::CannotGetBlockHash {
block_number: block_num,
reason: e
})?
.ok_or(OnlineClientAtBlockError::BlockNotFound {
block_number: block_num
})?;
(block_ref, block_num)
}
};
let mut historic_types = config.legacy_types_for_spec_version(spec_version);
// The metadata can be used to construct call and event types instead of us having to hardcode them all for every spec version:
let types_from_metadata = frame_decode::helpers::type_registry_from_metadata_any(&metadata)
.map_err(
|parse_error| OnlineClientAtBlockError::CannotInjectMetadataTypes { parse_error },
)?;
historic_types.prepend(types_from_metadata);
// Obtain the spec version so that we know which metadata to use at this block.
let spec_version = match self.inner.config.spec_version_for_block_number(block_num) {
Some(version) => version,
None => {
let block_hash = block_ref.hash();
let spec_version_bytes = self
.inner
.backend
.call("Core_version", None, block_hash)
.await
.map_err(|e| OnlineClientAtBlockError::CannotGetSpecVersion {
block_hash: block_hash.into(),
reason: e
})?;
Ok(ClientAtBlock::new(OnlineClientAtBlock {
config,
historic_types,
metadata,
rpc_methods,
block_hash,
}))
#[derive(codec::Decode)]
struct SpecVersionHeader {
_spec_name: String,
_impl_name: String,
_authoring_version: u32,
spec_version: u32,
}
SpecVersionHeader::decode(&mut &spec_version_bytes[..])
.map_err(|e| OnlineClientAtBlockError::CannotDecodeSpecVersion {
block_hash: block_hash.into(),
reason: e,
})?
.spec_version
}
};
// Obtain the metadata for the block, allowing our config to cache it.
let metadata = match self.inner.config.metadata_for_spec_version(spec_version) {
Some(metadata) => metadata,
None => {
//self.inner.backend.
todo!()
}
};
todo!()
}
}
@@ -189,7 +227,7 @@ pub struct OnlineClientAtBlock<T: Config> {
metadata: Arc<Metadata>,
backend: Arc<dyn Backend<T>>,
hasher: T::Hasher,
block_hash: HashFor<T>,
block_ref: BlockRef<HashFor<T>>,
}
impl<T: Config> OnlineClientAtBlockT<T> for OnlineClientAtBlock<T> {
@@ -197,7 +235,7 @@ impl<T: Config> OnlineClientAtBlockT<T> for OnlineClientAtBlock<T> {
&*self.backend
}
fn block_hash(&self) -> HashFor<T> {
self.block_hash
self.block_ref.hash()
}
}
@@ -207,66 +245,15 @@ impl<T: Config> OfflineClientAtBlockT for OnlineClientAtBlock<T> {
}
}
async fn get_spec_version<T: Config>(
rpc_methods: &ChainHeadRpcMethods<RpcConfigFor<T>>,
block_hash: HashFor<T>,
) -> Result<u32, OnlineClientAtBlockError> {
use codec::Decode;
use subxt_rpcs::methods::chain_head::ArchiveCallResult;
// make a runtime call to get the version information. This is also a constant
// in the metadata and so we could fetch it from there to avoid the call, but it would be a
// bit more effort.
let spec_version_bytes = {
let call_res = rpc_methods
.archive_v1_call(block_hash.into(), "Core_version", &[])
.await
.map_err(|e| OnlineClientAtBlockError::CannotGetSpecVersion {
block_hash: block_hash.to_string(),
reason: format!("Error calling Core_version: {e}"),
})?;
match call_res {
ArchiveCallResult::Success(bytes) => bytes.0,
ArchiveCallResult::Error(e) => {
return Err(OnlineClientAtBlockError::CannotGetSpecVersion {
block_hash: block_hash.to_string(),
reason: format!("Core_version returned an error: {e}"),
});
}
}
};
// We only care about the spec version, so just decode enough of this version information
// to be able to pluck out what we want, and ignore the rest.
let spec_version = {
#[derive(codec::Decode)]
struct SpecVersionHeader {
_spec_name: String,
_impl_name: String,
_authoring_version: u32,
spec_version: u32,
}
SpecVersionHeader::decode(&mut &spec_version_bytes[..])
.map_err(|e| OnlineClientAtBlockError::CannotGetSpecVersion {
block_hash: block_hash.to_string(),
reason: format!("Error decoding Core_version response: {e}"),
})?
.spec_version
};
Ok(spec_version)
}
async fn get_metadata<T: Config>(
rpc_methods: &ChainHeadRpcMethods<RpcConfigFor<T>>,
backend: &dyn Backend<T>,
block_hash: HashFor<T>,
) -> Result<RuntimeMetadata, OnlineClientAtBlockError> {
// First, try to use the "modern" metadata APIs to get the most recent version we can.
let version_to_get = rpc_methods
.archive_v1_call(block_hash.into(), "Metadata_metadata_versions", &[])
let version_to_get = backend
.call("Metadata_metadata_versions", None, block_hash)
.await
.ok()
.and_then(|res| res.as_success())
.and_then(|res| <Vec<u32>>::decode(&mut &res[..]).ok())
.and_then(|versions| {
// We want to filter out the "unstable" version, which is represented by u32::MAX.
@@ -276,12 +263,8 @@ async fn get_metadata<T: Config>(
// We had success calling the above API, so we expect the "modern" metadata API to work.
if let Some(version_to_get) = version_to_get {
let version_bytes = version_to_get.encode();
let rpc_response = rpc_methods
.archive_v1_call(
block_hash.into(),
"Metadata_metadata_at_version",
&version_bytes,
)
let rpc_response = backend
.call("Metadata_metadata_at_version", Some(&version_bytes), block_hash)
.await
.map_err(|e| OnlineClientAtBlockError::CannotGetMetadata {
block_hash: block_hash.to_string(),
new/src/client/online_client/block_number_or_ref.rs
+41
View File
@@ -0,0 +1,41 @@
use crate::config::{ Config, HashFor, Hasher };
use crate::backend::BlockRef;
/// This represents either a block number or a reference
/// to a block, which is essentially a block hash.
pub enum BlockNumberOrRef<T: Config> {
/// A block number.
Number(u64),
/// A block ref / hash.
BlockRef(BlockRef<HashFor<T>>)
}
impl <T: Config> From<u32> for BlockNumberOrRef<T> {
fn from(value: u32) -> Self {
BlockNumberOrRef::Number(value.into())
}
}
impl <T: Config> From<u64> for BlockNumberOrRef<T> {
fn from(value: u64) -> Self {
BlockNumberOrRef::Number(value)
}
}
impl <T: Config> From<BlockRef<HashFor<T>>> for BlockNumberOrRef<T> {
fn from(block_ref: BlockRef<HashFor<T>>) -> Self {
BlockNumberOrRef::BlockRef(block_ref)
}
}
// Ideally we'd have `impl From<HashFor<T>> for BlockNumberOrRef<T>` but since our config
// could set _any_ hash type, this boils down to `impl From<H> for ..` which is too general.
// Thus, we target our current concrete hash type.
impl <T: Config> From<crate::config::substrate::H256> for BlockNumberOrRef<T>
where
<T::Hasher as Hasher>::Hash: From<crate::config::substrate::H256>
{
fn from(hash: crate::config::substrate::H256) -> Self {
BlockNumberOrRef::BlockRef(BlockRef::from_hash(hash.into()))
}
}
new/src/config.rs
+2 -2
View File
@@ -63,7 +63,7 @@ pub trait Config: Clone + Debug + Sized + Send + Sync + 'static {
///
/// The [`crate::client::OnlineClient`] will look this up on chain if it's not available here,
/// but the [`crate::client::OfflineClient`] will error if this is not available for the required block number.
fn spec_version_for_block_number(&self, _block_number: u32) -> Option<u32> {
fn spec_version_for_block_number(&self, _block_number: u64) -> Option<u32> {
None
}
@@ -170,5 +170,5 @@ pub trait Hasher: Debug + Clone + Send + Sync + 'static {
/// This represents the block header type used by a node.
pub trait Header: Sized + Encode + Decode + Debug + Sync + Send + DeserializeOwned + Clone {
/// Return the block number of this header.
fn number(&self) -> u32;
fn number(&self) -> u64;
}
new/src/config/polkadot.rs
+1 -1
View File
@@ -75,7 +75,7 @@ impl Config for PolkadotConfig {
self.0.legacy_types_for_spec_version(spec_version)
}
fn spec_version_for_block_number(&self, block_number: u32) -> Option<u32> {
fn spec_version_for_block_number(&self, block_number: u64) -> Option<u32> {
self.0.spec_version_for_block_number(block_number)
}
new/src/config/substrate.rs
+6 -6
View File
@@ -20,7 +20,7 @@ use std::sync::Mutex;
/// Construct a [`SubstrateConfig`] using this.
pub struct SubstrateConfigBuilder {
legacy_types: Option<ChainTypeRegistry>,
spec_version_for_block_number: RangeMap<u32, u32>,
spec_version_for_block_number: RangeMap<u64, u32>,
metadata_for_spec_version: Mutex<HashMap<u32, Arc<Metadata>>>,
use_old_v9_hashers_before_spec_version: u32,
}
@@ -107,7 +107,7 @@ impl SubstrateConfigBuilder {
pub struct SpecVersionForRange {
/// The block range that this spec version applies to. Inclusive of the start
/// and exclusive of the enc.
pub block_range: std::ops::Range<u32>,
pub block_range: std::ops::Range<u64>,
/// The spec version at this block range.
pub spec_version: u32,
}
@@ -122,7 +122,7 @@ pub struct SubstrateConfig {
#[derive(Debug)]
struct SubstrateConfigInner {
legacy_types: Option<ChainTypeRegistry>,
spec_version_for_block_number: RangeMap<u32, u32>,
spec_version_for_block_number: RangeMap<u64, u32>,
metadata_for_spec_version: Mutex<HashMap<u32, Arc<Metadata>>>,
}
@@ -149,7 +149,7 @@ impl Config for SubstrateConfig {
.map(|types| types.for_spec_version(spec_version as u64))
}
fn spec_version_for_block_number(&self, block_number: u32) -> Option<u32> {
fn spec_version_for_block_number(&self, block_number: u64) -> Option<u32> {
self.inner.spec_version_for_block_number
.get(block_number)
.copied()
@@ -268,7 +268,7 @@ pub struct SubstrateHeader<Hash> {
deserialize_with = "deserialize_number"
)]
#[codec(compact)]
pub number: u32,
pub number: u64,
/// The state trie merkle root
pub state_root: Hash,
/// The merkle root of the extrinsics.
@@ -282,7 +282,7 @@ where
H: Hash,
SubstrateHeader<H>: Encode + Decode,
{
fn number(&self) -> u32 {
fn number(&self) -> u64 {
self.number.into()
}
}
new/src/error.rs
+49 -20
View File
@@ -171,7 +171,7 @@ pub enum OfflineClientAtBlockError {
)]
SpecVersionNotFound {
/// The block number for which the spec version was not found.
block_number: u32,
block_number: u64,
},
#[error(
"Cannot construct OfflineClientAtBlock: metadata not found for spec version {spec_version}"
@@ -193,6 +193,8 @@ pub enum OnlineClientError {
},
#[error("Cannot construct OnlineClient: {0}")]
RpcError(#[from] subxt_rpcs::Error),
#[error("Could not construct the CombinedBackend: {0}")]
CannotBuildCombinedBackend(CombinedBackendError),
#[error(
"Cannot construct OnlineClient: Cannot fetch latest finalized block to obtain init details from: {0}"
)]
@@ -229,38 +231,57 @@ pub enum OnlineClientAtBlockError {
/// Block number we failed to get the hash for.
block_number: u64,
/// The error we encountered.
reason: subxt_rpcs::Error,
reason: BackendError,
},
#[error("Cannot construct OnlineClientAtBlock: block number {block_number} not found")]
BlockNotFound {
/// The block number for which a block was not found.
block_number: u64,
},
#[error(
"Cannot construct OnlineClientAtBlock: failed to get spec version for block hash {block_hash}: {reason}"
)]
#[error("Cannot construct OnlineClientAtBlock: cannot get the block header for block {block_hash}: {reason}")]
CannotGetBlockHeader {
/// Block hash that we failed to fetch the header for.
block_hash: Hex,
/// The error we encountered.
reason: BackendError,
},
#[error("Cannot construct OnlineClientAtBlock: cannot find the block header for block {block_hash}")]
BlockHeaderNotFound {
/// Block hash that we failed to find the header for.
block_hash: Hex,
},
#[error("Cannot construct OnlineClientAtBlock: failed to obtain spec version for block {block_hash}: {reason}")]
CannotGetSpecVersion {
/// The block hash for which we failed to get the spec version.
block_hash: String,
/// The block hash for which we failed to obtain the spec version.
block_hash: Hex,
/// The error we encountered.
reason: String,
reason: BackendError,
},
#[error(
"Cannot construct OnlineClientAtBlock: failed to get metadata for block hash {block_hash}: {reason}"
"Cannot construct OnlineClientAtBlock: failed to decode spec version for block {block_hash}: {reason}"
)]
CannotGetMetadata {
/// The block hash for which we failed to get the metadata.
block_hash: String,
CannotDecodeSpecVersion {
/// The block hash for which we failed to decode the spec version.
block_hash: Hex,
/// The error we encountered.
reason: String,
},
#[error(
"Cannot inject types from metadata: failure to parse a type found in the metadata: {parse_error}"
)]
CannotInjectMetadataTypes {
/// Error parsing a type found in the metadata.
parse_error: scale_info_legacy::lookup_name::ParseError,
reason: codec::Error,
},
// #[error(
// "Cannot construct OnlineClientAtBlock: failed to get metadata for block {block_hash}: {reason}"
// )]
// CannotGetMetadata {
// /// The block hash for which we failed to get the metadata.
// block_hash: Hex,
// /// The error we encountered.
// reason: String,
// },
// #[error(
// "Cannot construct OnlineClientAtBlock: cannot inject types from metadata: failure to parse a type found in the metadata: {parse_error}"
// )]
// CannotInjectMetadataTypes {
// /// Error parsing a type found in the metadata.
// parse_error: scale_info_legacy::lookup_name::ParseError,
// },
}
#[derive(Debug, thiserror::Error)]
@@ -308,6 +329,14 @@ impl From<subxt_rpcs::Error> for BackendError {
}
}
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
#[allow(missing_docs)]
pub enum CombinedBackendError {
#[error("Could not obtain the list of RPC methods to determine which backends can be used")]
CouldNotObtainRpcMethodList(subxt_rpcs::Error)
}
/// An RPC error. Since we are generic over the RPC client that is used,
/// the error is boxed and could be casted.
#[derive(Debug, thiserror::Error)]