mirror of
https://github.com/pezkuwichain/pezkuwi-subxt.git
synced 2026-10-20 18:56:19 +00:00
light-client: Add experimental light-client support (#965)
* rpc/types: Decode `SubstrateTxStatus` for substrate and smoldot Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * lightclient: Add light client Error Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * lightclient: Add background task to manage RPC responses Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * lightclient: Implement the light client RPC in subxt Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * subxt: Expose light client under experimental feature-flag Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * artifacts: Add development chain spec for local nodes Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Update cargo lock Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * examples: Add light client example Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Update sp-* crates and smoldot to use git with branch / rev Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Apply cargo fmt Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Fix clippy Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Import hashmap entry Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * lightclient: Fetch spec only if jsonrpsee feature is enabled Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Update subxt/src/rpc/lightclient/background.rs Co-authored-by: Niklas Adolfsson <niklasadolfsson1@gmail.com> * Fix typo Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * artifacts: Update dev chain spec Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * types: Handle storage replies from chainHead_storage Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * artifacts: Add polkadot spec Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * lightclient: Handle RPC error responses Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * examples: Tx basic with light client for local nodes Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * example: Light client coprehensive example for live chains Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * examples: Remove prior light client example Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * feature: Rename experimental to unstable Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * book: Add light client section Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * testing: Fix clippy Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * lightclient: Ignore validated events Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Adjust tests for light-clients and normal clients Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * testing: Keep lightclient variant Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Remove support for chainHead_storage for light client Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Update light client to point to crates.io Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Update sp-crates from crates.io Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Replace Atomic with u64 Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Add LightClientBuilder Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Adjust chainspec with provided bootnodes Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Add potential_relay_chains to light client builder Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Move the light-client to the background task Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Adjust tracing logs Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Update book and example Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Apply cargo fmt Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Remove dev_spec.json artifact Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Examples fix duplicate Cargo.toml Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Use tracing_subscriber crate Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Fix clippy for different features Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Add comment about bootNodes Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Add comment about tracing-sub dependency Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Run integration-tests with light-client Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Feature guard some incompatible tests Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * ci: Enable light-client tests under feature flag Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * ci: Fix git step name Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Adjust flags for testing Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Adjust warnings Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Rename feature flag jsonrpsee-ws to jsonrpsee Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Fix cargo check Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * ci: Run tests on just 2 threads Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Move light-client to subxt/src/client Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Adjust LightClientBuilder Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Use ws_url to construct light client for testing Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Refactor background Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Address feedback Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Remove polkadot.spec and trim sub_id Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Wait for substrate to produce block before connecting light client Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Adjust builder and tests Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Apply fmt Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * ci: Use release for light client testing Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Add single test for light-client Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Wait for more blocks Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Use polkadot endpoint for testing Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Adjust cargo check Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * examples: Remove light client chain connection example Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Adjust cargo.toml section for the old example Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Adjust background task to use usize for subscription Id Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Build bootnodes with serde_json::Value directly Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Make channel between subxt user and subxt background unbounded Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Update subxt/src/client/lightclient/builder.rs Co-authored-by: Niklas Adolfsson <niklasadolfsson1@gmail.com> * Switch to smoldot 0.6.0 from 0.5.0 Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Move testing to `full_client` and `light_client` higher modules Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Remove subscriptionID type Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Remove subxt/integration-testing feature flag Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Adjust wait_for_blocks documentation Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Adjust utils import for testing Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> * Remove into_iter from builder construction Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> --------- Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io> Co-authored-by: Niklas Adolfsson <niklasadolfsson1@gmail.com>
This commit is contained in:
Alexandru VasileandNiklas Adolfsson authored and GitHub committed
1 parent
8413c4d2dd
commit
ef89752904
42 files changed
+2352
-147
No files matched your search
@@ -166,6 +166,38 @@ jobs:
|
||||
command: nextest
|
||||
args: run --workspace
|
||||
|
||||
nonwasm_light_client_tests:
|
||||
name: "Test Light Client"
|
||||
runs-on: ubuntu-latest-16-cores
|
||||
timeout-minutes: 15
|
||||
steps:
|
||||
- name: Checkout sources
|
||||
uses: actions/checkout@v3
|
||||
|
||||
- name: Download Substrate
|
||||
run: |
|
||||
curl $SUBSTRATE_URL --output substrate --location
|
||||
chmod +x substrate
|
||||
./substrate --version
|
||||
mkdir -p ~/.local/bin
|
||||
mv substrate ~/.local/bin
|
||||
|
||||
- name: Install Rust stable toolchain
|
||||
uses: actions-rs/toolchain@v1
|
||||
with:
|
||||
profile: minimal
|
||||
toolchain: stable
|
||||
override: true
|
||||
|
||||
- name: Rust Cache
|
||||
uses: Swatinem/rust-cache@988c164c3d0e93c4dbab36aaf5bbeb77425b2894 # v2.4.0
|
||||
|
||||
- name: Run tests
|
||||
uses: actions-rs/cargo@v1.0.3
|
||||
with:
|
||||
command: test
|
||||
args: --release --package integration-tests --features unstable-light-client
|
||||
|
||||
clippy:
|
||||
name: Cargo clippy
|
||||
runs-on: ubuntu-latest
|
||||
|
||||
Generated
+873
-104
File diff suppressed because it is too large.
Load diff
+6
-1
@@ -76,11 +76,16 @@ wabt = "0.10.0"
|
||||
wasm-bindgen-test = "0.3.24"
|
||||
which = "4.4.0"
|
||||
|
||||
# Light client support:
|
||||
smoldot-light = { version = "0.6.0", default-features = false }
|
||||
tokio-stream = "0.1.14"
|
||||
futures-util = "0.3.28"
|
||||
|
||||
# Substrate crates:
|
||||
sp-core = { version = "21.0.0", default-features = false }
|
||||
sp-core-hashing = "9.0.0"
|
||||
sp-keyring = "24.0.0"
|
||||
sp-runtime = "24.0.0"
|
||||
sp-keyring = "24.0.0"
|
||||
sp-version = "22.0.0"
|
||||
|
||||
# Subxt workspace crates:
|
||||
|
||||
+31
-5
@@ -53,10 +53,15 @@ substrate-compat = [
|
||||
# latest features exposed by the metadata.
|
||||
unstable-metadata = []
|
||||
|
||||
# Enable this to expose functionality only used for integration testing.
|
||||
# The exposed functionality is subject to breaking changes at any point,
|
||||
# and should not be relied upon.
|
||||
integration-tests = []
|
||||
# Activate this to expose the Light Client functionality.
|
||||
# Note that this feature is experimental and things may break or not work as expected.
|
||||
unstable-light-client = [
|
||||
"smoldot-light/std",
|
||||
"tokio-stream",
|
||||
"tokio/sync",
|
||||
"tokio/rt",
|
||||
"futures-util",
|
||||
]
|
||||
|
||||
[dependencies]
|
||||
codec = { package = "parity-scale-codec", workspace = true, features = ["derive"] }
|
||||
@@ -95,6 +100,12 @@ sp-runtime = { workspace = true, optional = true }
|
||||
subxt-macro = { workspace = true }
|
||||
subxt-metadata = { workspace = true }
|
||||
|
||||
# Light client support:
|
||||
smoldot-light = { workspace = true, optional = true }
|
||||
tokio = { workspace = true, optional = true }
|
||||
tokio-stream = { workspace = true, optional = true }
|
||||
futures-util = { workspace = true, optional = true }
|
||||
|
||||
# Included if "web" feature is enabled, to enable its js feature.
|
||||
getrandom = { workspace = true, optional = true }
|
||||
|
||||
@@ -108,4 +119,19 @@ sp-runtime = { workspace = true }
|
||||
sp-keyring = { workspace = true }
|
||||
sp-version = { workspace = true }
|
||||
assert_matches = { workspace = true }
|
||||
subxt-signer = { workspace = true, features = ["subxt"] }
|
||||
subxt-signer = { workspace = true, features = ["subxt"] }
|
||||
# Tracing subscriber is useful for light-client examples to ensure that
|
||||
# the `bootNodes` and chain spec are configured correctly. If all is fine, then
|
||||
# the light-client wlll emit INFO logs with
|
||||
# `GrandPa warp sync finished` and `Finalized block runtime ready.`
|
||||
tracing-subscriber = { workspace = true }
|
||||
|
||||
[[example]]
|
||||
name = "unstable_light_client_tx_basic"
|
||||
path = "examples/unstable_light_client_tx_basic.rs"
|
||||
required-features = ["unstable-light-client", "jsonrpsee"]
|
||||
|
||||
[profile.dev.package.smoldot-light]
|
||||
opt-level = 2
|
||||
[profile.test.package.smoldot-light]
|
||||
opt-level = 2
|
||||
@@ -12,7 +12,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
let storage_query = polkadot::storage().system().account_root();
|
||||
|
||||
// Get back an iterator of results (here, we are fetching 10 items at
|
||||
// a time from the node, but we always iterate over oen at a time).
|
||||
// a time from the node, but we always iterate over one at a time).
|
||||
let mut results = api
|
||||
.storage()
|
||||
.at_latest()
|
||||
|
||||
@@ -0,0 +1,51 @@
|
||||
use sp_keyring::AccountKeyring;
|
||||
use subxt::{
|
||||
client::{LightClient, LightClientBuilder, OfflineClientT},
|
||||
tx::PairSigner,
|
||||
PolkadotConfig,
|
||||
};
|
||||
|
||||
// Generate an interface that we can use from the node's metadata.
|
||||
#[subxt::subxt(runtime_metadata_path = "../artifacts/polkadot_metadata_small.scale")]
|
||||
pub mod polkadot {}
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
tracing_subscriber::fmt::init();
|
||||
|
||||
// Create a light client by fetching the chain spec of a local running node.
|
||||
// In this case, because we start one single node, the bootnodes must be overwritten
|
||||
// for the light client to connect to the local node.
|
||||
//
|
||||
// The `12D3KooWEyoppNCUx8Yx66oV9fJnriXwCcXwDDUA2kj6vnc6iDEp` is the P2P address
|
||||
// from a local polkadot node starting with
|
||||
// `--node-key 0000000000000000000000000000000000000000000000000000000000000001`
|
||||
let api: LightClient<PolkadotConfig> = LightClientBuilder::new()
|
||||
.bootnodes([
|
||||
"/ip4/127.0.0.1/tcp/30333/p2p/12D3KooWEyoppNCUx8Yx66oV9fJnriXwCcXwDDUA2kj6vnc6iDEp",
|
||||
])
|
||||
.build_from_url("ws://127.0.0.1:9944")
|
||||
.await?;
|
||||
|
||||
// Build a balance transfer extrinsic.
|
||||
let dest = AccountKeyring::Bob.to_account_id().into();
|
||||
let balance_transfer_tx = polkadot::tx().balances().transfer(dest, 10_000);
|
||||
|
||||
// Submit the balance transfer extrinsic from Alice, and wait for it to be successful
|
||||
// and in a finalized block. We get back the extrinsic events if all is well.
|
||||
let from = PairSigner::new(AccountKeyring::Alice.pair());
|
||||
let events = api
|
||||
.tx()
|
||||
.sign_and_submit_then_watch_default(&balance_transfer_tx, &from)
|
||||
.await?
|
||||
.wait_for_finalized_success()
|
||||
.await?;
|
||||
|
||||
// Find a Transfer event and print it.
|
||||
let transfer_event = events.find_first::<polkadot::balances::events::Transfer>()?;
|
||||
if let Some(event) = transfer_event {
|
||||
println!("Balance transfer success: {event:?}");
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
@@ -0,0 +1,48 @@
|
||||
// Copyright 2019-2023 Parity Technologies (UK) Ltd.
|
||||
// This file is dual-licensed as Apache-2.0 or GPL-3.0.
|
||||
// see LICENSE for license details.
|
||||
|
||||
//! # Light Client
|
||||
//!
|
||||
//! The Light Client aims to contribute to the decentralization of blockchains by providing connectivity
|
||||
//! to the P2P network and behaving similarly to a full node.
|
||||
//!
|
||||
//! To enable this functionality, the unstable-light-client feature flag needs to be enabled.
|
||||
//!
|
||||
//! To connect to a blockchain network, the Light Client requires a trusted sync state of the network, named "chain spec".
|
||||
//! This can be obtained by making a `sync_state_genSyncSpec` RPC call to a trusted node.
|
||||
//!
|
||||
//! The following is an example of fetching the chain spec from a local running onde on port 9933.
|
||||
//!
|
||||
//! ```bash
|
||||
//! curl -H "Content-Type: application/json" -d '{"id":1, "jsonrpc":"2.0", "method": "sync_state_genSyncSpec", "params":[true]}' http://localhost:9933/ | jq .result > chain_spec.json
|
||||
//! ```
|
||||
//!
|
||||
//! ## Example
|
||||
//!
|
||||
//! You can construct a Light Client from a trusted chain spec stored on disk.
|
||||
//! Similary, the Light Client can fetch the chain spec from a running node and
|
||||
//! overwrite the bootNodes section. The `jsonrpsee` feature flag exposes the
|
||||
//! `build_from_url` method.
|
||||
//!
|
||||
//! ```rust,ignore
|
||||
//! let light_client = LightClientBuilder::new()
|
||||
//! .bootnodes(
|
||||
//! ["/ip4/127.0.0.1/tcp/30333/p2p/12D3KooWEyoppNCUx8Yx66oV9fJnriXwCcXwDDUA2kj6vnc6iDEp"]
|
||||
//! )
|
||||
//! .build_from_url("ws://127.0.0.1:9944")
|
||||
//! .await?;
|
||||
//! ```
|
||||
//!
|
||||
//! Here's an example which connects to a local chain and submits a transaction.
|
||||
//!
|
||||
//! You can run the example using the following command:
|
||||
//!
|
||||
//! ```bash
|
||||
//! cargo run --example unstable_light_client_tx_basic --features="unstable-light-client jsonrpsee"
|
||||
//! ```
|
||||
//!
|
||||
//! ```rust,ignore
|
||||
#![doc = include_str!("../../../examples/unstable_light_client_tx_basic.rs")]
|
||||
//! ```
|
||||
//!
|
||||
@@ -10,12 +10,14 @@
|
||||
//! - [Constants](constants)
|
||||
//! - [Blocks](blocks)
|
||||
//! - [Runtime APIs](runtime_apis)
|
||||
//! - [Unstable Light Client](light_client)
|
||||
//!
|
||||
//! Alternately, [go back](super).
|
||||
|
||||
pub mod blocks;
|
||||
pub mod constants;
|
||||
pub mod events;
|
||||
pub mod light_client;
|
||||
pub mod runtime_apis;
|
||||
pub mod storage;
|
||||
pub mod transactions;
|
||||
@@ -0,0 +1,436 @@
|
||||
// Copyright 2019-2023 Parity Technologies (UK) Ltd.
|
||||
// This file is dual-licensed as Apache-2.0 or GPL-3.0.
|
||||
// see LICENSE for license details.
|
||||
|
||||
use futures::stream::StreamExt;
|
||||
use futures_util::future::{self, Either};
|
||||
use serde::Deserialize;
|
||||
use serde_json::value::RawValue;
|
||||
use std::{collections::HashMap, str::FromStr, sync::Arc};
|
||||
use tokio::sync::{mpsc, oneshot};
|
||||
|
||||
use super::LightClientError;
|
||||
use smoldot_light::{platform::default::DefaultPlatform as Platform, ChainId};
|
||||
|
||||
const LOG_TARGET: &str = "light-client-background";
|
||||
|
||||
/// The response of an RPC method.
|
||||
pub type MethodResponse = Result<Box<RawValue>, LightClientError>;
|
||||
|
||||
/// Message protocol between the front-end client that submits the RPC requests
|
||||
/// and the backend handler that produces responses from the chain.
|
||||
///
|
||||
/// The light client uses a single object [`smoldot_light::JsonRpcResponses`] to
|
||||
/// handle all requests and subscriptions from a chain. A background task is spawned
|
||||
/// to multiplex the rpc responses and to provide them back to their rightful submitters.
|
||||
#[derive(Debug)]
|
||||
pub enum FromSubxt {
|
||||
/// The RPC method request.
|
||||
Request {
|
||||
/// The method of the request.
|
||||
method: String,
|
||||
/// The parameters of the request.
|
||||
params: String,
|
||||
/// Channel used to send back the result.
|
||||
sender: oneshot::Sender<MethodResponse>,
|
||||
},
|
||||
/// The RPC subscription (pub/sub) request.
|
||||
Subscription {
|
||||
/// The method of the request.
|
||||
method: String,
|
||||
/// The parameters of the request.
|
||||
params: String,
|
||||
/// Channel used to send back the subscription ID if successful.
|
||||
sub_id: oneshot::Sender<MethodResponse>,
|
||||
/// Channel used to send back the notifcations.
|
||||
sender: mpsc::UnboundedSender<Box<RawValue>>,
|
||||
},
|
||||
}
|
||||
|
||||
/// Background task data.
|
||||
pub struct BackgroundTask {
|
||||
/// Smoldot light client implementation that leverages the exposed platform.
|
||||
client: smoldot_light::Client<Arc<Platform>>,
|
||||
/// The ID of the chain used to identify the chain protocol (ie. substrate).
|
||||
///
|
||||
/// Note: A single chain is supported for a client. This aligns with the subxt's
|
||||
/// vision of the Client.
|
||||
chain_id: ChainId,
|
||||
/// Unique ID for RPC calls.
|
||||
request_id: usize,
|
||||
/// Map the request ID of a RPC method to the frontend `Sender`.
|
||||
requests: HashMap<usize, oneshot::Sender<MethodResponse>>,
|
||||
/// Subscription calls first need to make a plain RPC method
|
||||
/// request to obtain the subscription ID.
|
||||
///
|
||||
/// The RPC method request is made in the background and the response should
|
||||
/// not be sent back to the user.
|
||||
/// Map the request ID of a RPC method to the frontend `Sender`.
|
||||
id_to_subscription: HashMap<
|
||||
usize,
|
||||
(
|
||||
oneshot::Sender<MethodResponse>,
|
||||
mpsc::UnboundedSender<Box<RawValue>>,
|
||||
),
|
||||
>,
|
||||
/// Map the subscription ID to the frontend `Sender`.
|
||||
subscriptions: HashMap<usize, mpsc::UnboundedSender<Box<RawValue>>>,
|
||||
}
|
||||
|
||||
impl BackgroundTask {
|
||||
/// Constructs a new [`BackgroundTask`].
|
||||
pub fn new(client: smoldot_light::Client<Arc<Platform>>, chain_id: ChainId) -> BackgroundTask {
|
||||
BackgroundTask {
|
||||
client,
|
||||
chain_id,
|
||||
request_id: 1,
|
||||
requests: Default::default(),
|
||||
id_to_subscription: Default::default(),
|
||||
subscriptions: Default::default(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Fetch and increment the request ID.
|
||||
fn next_id(&mut self) -> usize {
|
||||
let next = self.request_id;
|
||||
self.request_id = self.request_id.wrapping_add(1);
|
||||
next
|
||||
}
|
||||
|
||||
/// Handle the registration messages received from the user.
|
||||
async fn handle_requests(&mut self, message: FromSubxt) {
|
||||
match message {
|
||||
FromSubxt::Request {
|
||||
method,
|
||||
params,
|
||||
sender,
|
||||
} => {
|
||||
let id = self.next_id();
|
||||
let request = format!(
|
||||
r#"{{"jsonrpc":"2.0","id":"{}", "method":"{}","params":{}}}"#,
|
||||
id, method, params
|
||||
);
|
||||
|
||||
self.requests.insert(id, sender);
|
||||
|
||||
let result = self.client.json_rpc_request(request, self.chain_id);
|
||||
if let Err(err) = result {
|
||||
tracing::warn!(
|
||||
target: LOG_TARGET,
|
||||
"Cannot send RPC request to lightclient {:?}",
|
||||
err.to_string()
|
||||
);
|
||||
let sender = self
|
||||
.requests
|
||||
.remove(&id)
|
||||
.expect("Channel is inserted above; qed");
|
||||
|
||||
// Send the error back to frontend.
|
||||
if sender
|
||||
.send(Err(LightClientError::Request(err.to_string())))
|
||||
.is_err()
|
||||
{
|
||||
tracing::warn!(
|
||||
target: LOG_TARGET,
|
||||
"Cannot send RPC request error to id={id}",
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
FromSubxt::Subscription {
|
||||
method,
|
||||
params,
|
||||
sub_id,
|
||||
sender,
|
||||
} => {
|
||||
// For subscriptions we need to make a plain RPC request to the subscription method.
|
||||
// The server will return as a result the subscription ID.
|
||||
let id = self.next_id();
|
||||
let request = format!(
|
||||
r#"{{"jsonrpc":"2.0","id":"{}", "method":"{}","params":{}}}"#,
|
||||
id, method, params
|
||||
);
|
||||
|
||||
self.id_to_subscription.insert(id, (sub_id, sender));
|
||||
|
||||
let result = self.client.json_rpc_request(request, self.chain_id);
|
||||
if let Err(err) = result {
|
||||
tracing::warn!(
|
||||
target: LOG_TARGET,
|
||||
"Cannot send RPC request to lightclient {:?}",
|
||||
err.to_string()
|
||||
);
|
||||
let (sub_id, _) = self
|
||||
.id_to_subscription
|
||||
.remove(&id)
|
||||
.expect("Channels are inserted above; qed");
|
||||
|
||||
// Send the error back to frontend.
|
||||
if sub_id
|
||||
.send(Err(LightClientError::Request(err.to_string())))
|
||||
.is_err()
|
||||
{
|
||||
tracing::warn!(
|
||||
target: LOG_TARGET,
|
||||
"Cannot send RPC request error to id={id}",
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
/// Parse the response received from the light client and sent it to the appropriate user.
|
||||
fn handle_rpc_response(&mut self, response: String) {
|
||||
match RpcResponse::from_str(&response) {
|
||||
Ok(RpcResponse::Error { id, error }) => {
|
||||
let Ok(id) = id.parse::<usize>() else {
|
||||
tracing::warn!(target: LOG_TARGET, "Cannot send error. Id={id} is not a valid number");
|
||||
return
|
||||
};
|
||||
|
||||
if let Some(sender) = self.requests.remove(&id) {
|
||||
if sender
|
||||
.send(Err(LightClientError::Request(error.to_string())))
|
||||
.is_err()
|
||||
{
|
||||
tracing::warn!(
|
||||
target: LOG_TARGET,
|
||||
"Cannot send method response to id={id}",
|
||||
);
|
||||
}
|
||||
} else if let Some((sub_id_sender, _)) = self.id_to_subscription.remove(&id) {
|
||||
if sub_id_sender
|
||||
.send(Err(LightClientError::Request(error.to_string())))
|
||||
.is_err()
|
||||
{
|
||||
tracing::warn!(
|
||||
target: LOG_TARGET,
|
||||
"Cannot send method response to id {:?}",
|
||||
id
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(RpcResponse::Method { id, result }) => {
|
||||
let Ok(id) = id.parse::<usize>() else {
|
||||
tracing::warn!(target: LOG_TARGET, "Cannot send response. Id={id} is not a valid number");
|
||||
return
|
||||
};
|
||||
|
||||
// Send the response back.
|
||||
if let Some(sender) = self.requests.remove(&id) {
|
||||
if sender.send(Ok(result)).is_err() {
|
||||
tracing::warn!(
|
||||
target: LOG_TARGET,
|
||||
"Cannot send method response to id={id}",
|
||||
);
|
||||
}
|
||||
} else if let Some((sub_id_sender, sender)) = self.id_to_subscription.remove(&id) {
|
||||
let Ok(sub_id) = result
|
||||
.get()
|
||||
.trim_start_matches('"')
|
||||
.trim_end_matches('"')
|
||||
.parse::<usize>() else {
|
||||
tracing::warn!(
|
||||
target: LOG_TARGET,
|
||||
"Subscription id={result} is not a valid number",
|
||||
);
|
||||
return;
|
||||
};
|
||||
|
||||
tracing::trace!(target: LOG_TARGET, "Received subscription id={sub_id}");
|
||||
|
||||
if sub_id_sender.send(Ok(result)).is_err() {
|
||||
tracing::warn!(
|
||||
target: LOG_TARGET,
|
||||
"Cannot send method response to id={id}",
|
||||
);
|
||||
} else {
|
||||
// Track this subscription ID if send is successful.
|
||||
self.subscriptions.insert(sub_id, sender);
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(RpcResponse::Subscription { method, id, result }) => {
|
||||
let Ok(id) = id.parse::<usize>() else {
|
||||
tracing::warn!(target: LOG_TARGET, "Cannot send subscription. Id={id} is not a valid number");
|
||||
return
|
||||
};
|
||||
|
||||
if let Some(sender) = self.subscriptions.get_mut(&id) {
|
||||
// Send the current notification response.
|
||||
if sender.send(result).is_err() {
|
||||
tracing::warn!(
|
||||
target: LOG_TARGET,
|
||||
"Cannot send notification to subscription id={id} method={method}",
|
||||
);
|
||||
|
||||
// Remove the sender if the subscription dropped the receiver.
|
||||
self.subscriptions.remove(&id);
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(err) => {
|
||||
tracing::warn!(target: LOG_TARGET, "cannot decode RPC response {:?}", err);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Perform the main background task:
|
||||
/// - receiving requests from subxt RPC method / subscriptions
|
||||
/// - provides the results from the light client back to users.
|
||||
pub async fn start_task(
|
||||
&mut self,
|
||||
from_subxt: mpsc::UnboundedReceiver<FromSubxt>,
|
||||
from_node: smoldot_light::JsonRpcResponses,
|
||||
) {
|
||||
let from_subxt_event = tokio_stream::wrappers::UnboundedReceiverStream::new(from_subxt);
|
||||
let from_node_event = futures_util::stream::unfold(from_node, |mut from_node| async {
|
||||
from_node.next().await.map(|result| (result, from_node))
|
||||
});
|
||||
|
||||
tokio::pin!(from_subxt_event, from_node_event);
|
||||
|
||||
let mut from_subxt_event_fut = from_subxt_event.next();
|
||||
let mut from_node_event_fut = from_node_event.next();
|
||||
|
||||
loop {
|
||||
match future::select(from_subxt_event_fut, from_node_event_fut).await {
|
||||
// Message received from subxt.
|
||||
Either::Left((subxt_message, previous_fut)) => {
|
||||
let Some(message) = subxt_message else {
|
||||
tracing::trace!(target: LOG_TARGET, "Subxt channel closed");
|
||||
break;
|
||||
};
|
||||
tracing::trace!(
|
||||
target: LOG_TARGET,
|
||||
"Received register message {:?}",
|
||||
message
|
||||
);
|
||||
|
||||
self.handle_requests(message).await;
|
||||
|
||||
from_subxt_event_fut = from_subxt_event.next();
|
||||
from_node_event_fut = previous_fut;
|
||||
}
|
||||
// Message received from rpc handler: lightclient response.
|
||||
Either::Right((node_message, previous_fut)) => {
|
||||
// Smoldot returns `None` if the chain has been removed (which subxt does not remove).
|
||||
let Some(response) = node_message else {
|
||||
tracing::trace!(target: LOG_TARGET, "Smoldot RPC responses channel closed");
|
||||
break;
|
||||
};
|
||||
tracing::trace!(
|
||||
target: LOG_TARGET,
|
||||
"Received smoldot RPC result {:?}",
|
||||
response
|
||||
);
|
||||
|
||||
self.handle_rpc_response(response);
|
||||
|
||||
// Advance backend, save frontend.
|
||||
from_subxt_event_fut = previous_fut;
|
||||
from_node_event_fut = from_node_event.next();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
tracing::trace!(target: LOG_TARGET, "Task closed");
|
||||
}
|
||||
}
|
||||
|
||||
/// The RPC response from the light-client.
|
||||
/// This can either be a response of a method, or a notification from a subscription.
|
||||
#[derive(Debug, Clone)]
|
||||
enum RpcResponse {
|
||||
Method {
|
||||
/// Response ID.
|
||||
id: String,
|
||||
/// The result of the method call.
|
||||
result: Box<RawValue>,
|
||||
},
|
||||
Subscription {
|
||||
/// RPC method that generated the notification.
|
||||
method: String,
|
||||
/// Subscription ID.
|
||||
id: String,
|
||||
/// Result.
|
||||
result: Box<RawValue>,
|
||||
},
|
||||
Error {
|
||||
/// Response ID.
|
||||
id: String,
|
||||
/// Error.
|
||||
error: Box<RawValue>,
|
||||
},
|
||||
}
|
||||
|
||||
impl std::str::FromStr for RpcResponse {
|
||||
type Err = serde_json::Error;
|
||||
|
||||
fn from_str(response: &str) -> Result<Self, Self::Err> {
|
||||
// Helper structures to deserialize from raw RPC strings.
|
||||
#[derive(Deserialize, Debug)]
|
||||
struct Response {
|
||||
/// JSON-RPC version.
|
||||
#[allow(unused)]
|
||||
jsonrpc: String,
|
||||
/// Result.
|
||||
result: Box<RawValue>,
|
||||
/// Request ID
|
||||
id: String,
|
||||
}
|
||||
#[derive(Deserialize)]
|
||||
struct NotificationParams {
|
||||
/// The ID of the subscription.
|
||||
subscription: String,
|
||||
/// Result.
|
||||
result: Box<RawValue>,
|
||||
}
|
||||
#[derive(Deserialize)]
|
||||
struct ResponseNotification {
|
||||
/// JSON-RPC version.
|
||||
#[allow(unused)]
|
||||
jsonrpc: String,
|
||||
/// RPC method that generated the notification.
|
||||
method: String,
|
||||
/// Result.
|
||||
params: NotificationParams,
|
||||
}
|
||||
#[derive(Deserialize)]
|
||||
struct ErrorResponse {
|
||||
/// JSON-RPC version.
|
||||
#[allow(unused)]
|
||||
jsonrpc: String,
|
||||
/// Request ID.
|
||||
id: String,
|
||||
/// Error.
|
||||
error: Box<RawValue>,
|
||||
}
|
||||
|
||||
// Check if the response can be mapped as an RPC method response.
|
||||
let result: Result<Response, _> = serde_json::from_str(response);
|
||||
if let Ok(response) = result {
|
||||
return Ok(RpcResponse::Method {
|
||||
id: response.id,
|
||||
result: response.result,
|
||||
});
|
||||
}
|
||||
|
||||
let result: Result<ResponseNotification, _> = serde_json::from_str(response);
|
||||
if let Ok(notification) = result {
|
||||
return Ok(RpcResponse::Subscription {
|
||||
id: notification.params.subscription,
|
||||
method: notification.method,
|
||||
result: notification.params.result,
|
||||
});
|
||||
}
|
||||
|
||||
let error: ErrorResponse = serde_json::from_str(response)?;
|
||||
Ok(RpcResponse::Error {
|
||||
id: error.id,
|
||||
error: error.error,
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,191 @@
|
||||
// Copyright 2019-2023 Parity Technologies (UK) Ltd.
|
||||
// This file is dual-licensed as Apache-2.0 or GPL-3.0.
|
||||
// see LICENSE for license details.
|
||||
|
||||
use super::{rpc::LightClientRpc, LightClient, LightClientError};
|
||||
use crate::{config::Config, error::Error, OnlineClient};
|
||||
|
||||
#[cfg(feature = "jsonrpsee")]
|
||||
use jsonrpsee::{
|
||||
async_client::ClientBuilder,
|
||||
client_transport::ws::{Uri, WsTransportClientBuilder},
|
||||
core::client::ClientT,
|
||||
rpc_params,
|
||||
};
|
||||
use smoldot_light::ChainId;
|
||||
use std::num::NonZeroU32;
|
||||
use std::sync::Arc;
|
||||
|
||||
/// Builder for [`LightClient`].
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct LightClientBuilder {
|
||||
max_pending_requests: NonZeroU32,
|
||||
max_subscriptions: u32,
|
||||
bootnodes: Option<Vec<serde_json::Value>>,
|
||||
potential_relay_chains: Option<Vec<ChainId>>,
|
||||
}
|
||||
|
||||
impl Default for LightClientBuilder {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
max_pending_requests: NonZeroU32::new(128)
|
||||
.expect("Valid number is greater than zero; qed"),
|
||||
max_subscriptions: 1024,
|
||||
bootnodes: None,
|
||||
potential_relay_chains: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl LightClientBuilder {
|
||||
/// Create a new [`LightClientBuilder`].
|
||||
pub fn new() -> LightClientBuilder {
|
||||
LightClientBuilder::default()
|
||||
}
|
||||
|
||||
/// Overwrite the bootnodes of the chain specification.
|
||||
///
|
||||
/// Can be used to provide trusted entities to the chain spec, or for
|
||||
/// testing environments.
|
||||
pub fn bootnodes<'a>(mut self, bootnodes: impl IntoIterator<Item = &'a str>) -> Self {
|
||||
self.bootnodes = Some(bootnodes.into_iter().map(Into::into).collect());
|
||||
self
|
||||
}
|
||||
|
||||
/// Maximum number of JSON-RPC in the queue of requests waiting to be processed.
|
||||
/// This parameter is necessary for situations where the JSON-RPC clients aren't
|
||||
/// trusted. If you control all the requests that are sent out and don't want them
|
||||
/// to fail, feel free to pass `u32::max_value()`.
|
||||
///
|
||||
/// Default is 128.
|
||||
pub fn max_pending_requests(mut self, max_pending_requests: NonZeroU32) -> Self {
|
||||
self.max_pending_requests = max_pending_requests;
|
||||
self
|
||||
}
|
||||
|
||||
/// Maximum number of active subscriptions before new ones are automatically
|
||||
/// rejected. Any JSON-RPC request that causes the server to generate notifications
|
||||
/// counts as a subscription.
|
||||
///
|
||||
/// Default is 1024.
|
||||
pub fn max_subscriptions(mut self, max_subscriptions: u32) -> Self {
|
||||
self.max_subscriptions = max_subscriptions;
|
||||
self
|
||||
}
|
||||
|
||||
/// If the chain spec defines a parachain, contains the list of relay chains to choose
|
||||
/// from. Ignored if not a parachain.
|
||||
///
|
||||
/// This field is necessary because multiple different chain can have the same identity.
|
||||
///
|
||||
/// For example: if user A adds a chain named "Kusama", then user B adds a different chain
|
||||
/// also named "Kusama", then user B adds a parachain whose relay chain is "Kusama", it would
|
||||
/// be wrong to connect to the "Kusama" created by user A.
|
||||
pub fn potential_relay_chains(
|
||||
mut self,
|
||||
potential_relay_chains: impl IntoIterator<Item = ChainId>,
|
||||
) -> Self {
|
||||
self.potential_relay_chains = Some(potential_relay_chains.into_iter().collect());
|
||||
self
|
||||
}
|
||||
|
||||
/// Build the light client with specified URL to connect to.
|
||||
/// You must provide the port number in the URL.
|
||||
///
|
||||
/// ## Panics
|
||||
///
|
||||
/// Panics if being called outside of `tokio` runtime context.
|
||||
#[cfg(feature = "jsonrpsee")]
|
||||
pub async fn build_from_url<T: Config, Url: AsRef<str>>(
|
||||
self,
|
||||
url: Url,
|
||||
) -> Result<LightClient<T>, Error> {
|
||||
let chain_spec = fetch_url(url.as_ref()).await?;
|
||||
|
||||
self.build_client(chain_spec).await
|
||||
}
|
||||
|
||||
/// Build the light client from chain spec.
|
||||
///
|
||||
/// The most important field of the configuration is the chain specification.
|
||||
/// This is a JSON document containing all the information necessary for the client to
|
||||
/// connect to said chain.
|
||||
///
|
||||
/// The chain spec must be obtained from a trusted entity.
|
||||
///
|
||||
/// It can be fetched from a trused node with the following command:
|
||||
/// ```bash
|
||||
/// curl -H "Content-Type: application/json" -d '{"id":1, "jsonrpc":"2.0", "method": "sync_state_genSyncSpec", "params":[true]}' http://localhost:9944/ | jq .result > res.spec
|
||||
/// ```
|
||||
///
|
||||
/// # Note
|
||||
///
|
||||
/// For testing environments, please populate the "bootNodes" if the not already provided.
|
||||
/// See [`Self::bootnodes`] for more details.
|
||||
///
|
||||
/// ## Panics
|
||||
///
|
||||
/// Panics if being called outside of `tokio` runtime context.
|
||||
pub async fn build<T: Config>(self, chain_spec: &str) -> Result<LightClient<T>, Error> {
|
||||
let chain_spec = serde_json::from_str(chain_spec)
|
||||
.map_err(|_| Error::LightClient(LightClientError::InvalidChainSpec))?;
|
||||
|
||||
self.build_client(chain_spec).await
|
||||
}
|
||||
|
||||
/// Build the light client.
|
||||
async fn build_client<T: Config>(
|
||||
self,
|
||||
mut chain_spec: serde_json::Value,
|
||||
) -> Result<LightClient<T>, Error> {
|
||||
// Set custom bootnodes if provided.
|
||||
if let Some(bootnodes) = self.bootnodes {
|
||||
if let serde_json::Value::Object(map) = &mut chain_spec {
|
||||
map.insert("bootNodes".to_string(), serde_json::Value::Array(bootnodes));
|
||||
}
|
||||
}
|
||||
|
||||
let config = smoldot_light::AddChainConfig {
|
||||
specification: &chain_spec.to_string(),
|
||||
json_rpc: smoldot_light::AddChainConfigJsonRpc::Enabled {
|
||||
max_pending_requests: self.max_pending_requests,
|
||||
max_subscriptions: self.max_subscriptions,
|
||||
},
|
||||
potential_relay_chains: self.potential_relay_chains.unwrap_or_default().into_iter(),
|
||||
database_content: "",
|
||||
user_data: (),
|
||||
};
|
||||
|
||||
let rpc = LightClientRpc::new(config)?;
|
||||
let online_client = OnlineClient::<T>::from_rpc_client(Arc::new(rpc)).await?;
|
||||
Ok(LightClient(online_client))
|
||||
}
|
||||
}
|
||||
|
||||
/// Fetch the chain spec from the URL.
|
||||
#[cfg(feature = "jsonrpsee")]
|
||||
async fn fetch_url(url: impl AsRef<str>) -> Result<serde_json::Value, Error> {
|
||||
let url = url
|
||||
.as_ref()
|
||||
.parse::<Uri>()
|
||||
.map_err(|_| Error::LightClient(LightClientError::InvalidUrl))?;
|
||||
|
||||
if url.scheme_str() != Some("ws") && url.scheme_str() != Some("wss") {
|
||||
return Err(Error::LightClient(LightClientError::InvalidScheme));
|
||||
}
|
||||
|
||||
let (sender, receiver) = WsTransportClientBuilder::default()
|
||||
.build(url)
|
||||
.await
|
||||
.map_err(|_| LightClientError::Handshake)?;
|
||||
|
||||
let client = ClientBuilder::default()
|
||||
.request_timeout(core::time::Duration::from_secs(180))
|
||||
.max_notifs_per_subscription(4096)
|
||||
.build_with_tokio(sender, receiver);
|
||||
|
||||
client
|
||||
.request("sync_state_genSyncSpec", rpc_params![true])
|
||||
.await
|
||||
.map_err(|err| Error::Rpc(crate::error::RpcError::ClientError(Box::new(err))))
|
||||
}
|
||||
@@ -0,0 +1,75 @@
|
||||
// Copyright 2019-2023 Parity Technologies (UK) Ltd.
|
||||
// This file is dual-licensed as Apache-2.0 or GPL-3.0.
|
||||
// see LICENSE for license details.
|
||||
|
||||
//! This module provides support for light clients.
|
||||
|
||||
mod background;
|
||||
mod builder;
|
||||
mod rpc;
|
||||
|
||||
use derivative::Derivative;
|
||||
|
||||
use crate::{
|
||||
client::{OfflineClientT, OnlineClientT},
|
||||
config::Config,
|
||||
OnlineClient,
|
||||
};
|
||||
|
||||
pub use builder::LightClientBuilder;
|
||||
|
||||
/// Light client error.
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum LightClientError {
|
||||
/// Error encountered while adding the chain to the light-client.
|
||||
#[error("Failed to add the chain to the light client: {0}.")]
|
||||
AddChainError(String),
|
||||
/// The background task is closed.
|
||||
#[error("Failed to communicate with the background task.")]
|
||||
BackgroundClosed,
|
||||
/// Invalid RPC parameters cannot be serialized as JSON string.
|
||||
#[error("RPC parameters cannot be serialized as JSON string.")]
|
||||
InvalidParams,
|
||||
/// Error originated while trying to submit a RPC request.
|
||||
#[error("RPC request cannot be sent: {0}.")]
|
||||
Request(String),
|
||||
/// The provided URL scheme is invalid.
|
||||
///
|
||||
/// Supported versions: WS, WSS.
|
||||
#[error("The provided URL scheme is invalid.")]
|
||||
InvalidScheme,
|
||||
/// The provided URL is invalid.
|
||||
#[error("The provided URL scheme is invalid.")]
|
||||
InvalidUrl,
|
||||
/// The provided chain spec is invalid.
|
||||
#[error("The provided chain spec is not a valid JSON object.")]
|
||||
InvalidChainSpec,
|
||||
/// Handshake error while connecting to a node.
|
||||
#[error("WS handshake failed.")]
|
||||
Handshake,
|
||||
}
|
||||
|
||||
/// The light-client RPC implementation that is used to connect with the chain.
|
||||
#[derive(Derivative)]
|
||||
#[derivative(Clone(bound = ""))]
|
||||
pub struct LightClient<T: Config>(OnlineClient<T>);
|
||||
|
||||
impl<T: Config> OnlineClientT<T> for LightClient<T> {
|
||||
fn rpc(&self) -> &crate::rpc::Rpc<T> {
|
||||
self.0.rpc()
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: Config> OfflineClientT<T> for LightClient<T> {
|
||||
fn metadata(&self) -> crate::Metadata {
|
||||
self.0.metadata()
|
||||
}
|
||||
|
||||
fn genesis_hash(&self) -> <T as Config>::Hash {
|
||||
self.0.genesis_hash()
|
||||
}
|
||||
|
||||
fn runtime_version(&self) -> crate::rpc::types::RuntimeVersion {
|
||||
self.0.runtime_version()
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,209 @@
|
||||
// Copyright 2019-2023 Parity Technologies (UK) Ltd.
|
||||
// This file is dual-licensed as Apache-2.0 or GPL-3.0.
|
||||
// see LICENSE for license details.
|
||||
|
||||
use super::{
|
||||
background::{BackgroundTask, FromSubxt, MethodResponse},
|
||||
LightClientError,
|
||||
};
|
||||
use crate::{
|
||||
error::{Error, RpcError},
|
||||
rpc::{RpcClientT, RpcFuture, RpcSubscription},
|
||||
};
|
||||
use futures::{stream::StreamExt, Stream};
|
||||
use serde_json::value::RawValue;
|
||||
use smoldot_light::{platform::default::DefaultPlatform as Platform, ChainId};
|
||||
use std::pin::Pin;
|
||||
use tokio::sync::{mpsc, mpsc::error::SendError, oneshot};
|
||||
use tokio_stream::wrappers::UnboundedReceiverStream;
|
||||
|
||||
pub const LOG_TARGET: &str = "light-client";
|
||||
|
||||
/// The light-client RPC implementation that is used to connect with the chain.
|
||||
#[derive(Clone)]
|
||||
pub struct LightClientRpc {
|
||||
/// Communicate with the backend task that multiplexes the responses
|
||||
/// back to the frontend.
|
||||
to_backend: mpsc::UnboundedSender<FromSubxt>,
|
||||
}
|
||||
|
||||
impl LightClientRpc {
|
||||
/// Constructs a new [`LightClientRpc`], providing the chain specification.
|
||||
///
|
||||
/// The chain specification can be downloaded from a trusted network via
|
||||
/// the `sync_state_genSyncSpec` RPC method. This parameter expects the
|
||||
/// chain spec in text format (ie not in hex-encoded scale-encoded as RPC methods
|
||||
/// will provide).
|
||||
///
|
||||
/// ## Panics
|
||||
///
|
||||
/// Panics if being called outside of `tokio` runtime context.
|
||||
pub fn new(
|
||||
config: smoldot_light::AddChainConfig<'_, (), impl Iterator<Item = ChainId>>,
|
||||
) -> Result<LightClientRpc, Error> {
|
||||
tracing::trace!(target: LOG_TARGET, "Create light client");
|
||||
|
||||
let mut client = smoldot_light::Client::new(Platform::new(
|
||||
env!("CARGO_PKG_NAME").into(),
|
||||
env!("CARGO_PKG_VERSION").into(),
|
||||
));
|
||||
|
||||
let smoldot_light::AddChainSuccess {
|
||||
chain_id,
|
||||
json_rpc_responses,
|
||||
} = client
|
||||
.add_chain(config)
|
||||
.map_err(|err| LightClientError::AddChainError(err.to_string()))?;
|
||||
|
||||
let (to_backend, backend) = mpsc::unbounded_channel();
|
||||
|
||||
// `json_rpc_responses` can only be `None` if we had passed `json_rpc: Disabled`.
|
||||
let rpc_responses = json_rpc_responses.expect("Light client RPC configured; qed");
|
||||
|
||||
tokio::spawn(async move {
|
||||
let mut task = BackgroundTask::new(client, chain_id);
|
||||
task.start_task(backend, rpc_responses).await;
|
||||
});
|
||||
|
||||
Ok(LightClientRpc { to_backend })
|
||||
}
|
||||
|
||||
/// Submits an RPC method request to the light-client.
|
||||
///
|
||||
/// This method sends a request to the light-client to execute an RPC method with the provided parameters.
|
||||
/// The parameters are parsed into a valid JSON object in the background.
|
||||
fn method_request(
|
||||
&self,
|
||||
method: String,
|
||||
params: String,
|
||||
) -> Result<oneshot::Receiver<MethodResponse>, SendError<FromSubxt>> {
|
||||
let (sender, receiver) = oneshot::channel();
|
||||
|
||||
self.to_backend.send(FromSubxt::Request {
|
||||
method,
|
||||
params,
|
||||
sender,
|
||||
})?;
|
||||
|
||||
Ok(receiver)
|
||||
}
|
||||
|
||||
/// Makes an RPC subscription call to the light-client.
|
||||
///
|
||||
/// This method sends a request to the light-client to establish an RPC subscription with the provided parameters.
|
||||
/// The parameters are parsed into a valid JSON object in the background.
|
||||
fn subscription_request(
|
||||
&self,
|
||||
method: String,
|
||||
params: String,
|
||||
) -> Result<
|
||||
(
|
||||
oneshot::Receiver<MethodResponse>,
|
||||
mpsc::UnboundedReceiver<Box<RawValue>>,
|
||||
),
|
||||
SendError<FromSubxt>,
|
||||
> {
|
||||
let (sub_id, sub_id_rx) = oneshot::channel();
|
||||
let (sender, receiver) = mpsc::unbounded_channel();
|
||||
|
||||
self.to_backend.send(FromSubxt::Subscription {
|
||||
method,
|
||||
params,
|
||||
sub_id,
|
||||
sender,
|
||||
})?;
|
||||
|
||||
Ok((sub_id_rx, receiver))
|
||||
}
|
||||
}
|
||||
|
||||
impl RpcClientT for LightClientRpc {
|
||||
fn request_raw<'a>(
|
||||
&'a self,
|
||||
method: &'a str,
|
||||
params: Option<Box<RawValue>>,
|
||||
) -> RpcFuture<'a, Box<RawValue>> {
|
||||
let client = self.clone();
|
||||
|
||||
Box::pin(async move {
|
||||
let params = match params {
|
||||
Some(params) => serde_json::to_string(¶ms).map_err(|_| {
|
||||
RpcError::ClientError(Box::new(LightClientError::InvalidParams))
|
||||
})?,
|
||||
None => "[]".into(),
|
||||
};
|
||||
|
||||
// Fails if the background is closed.
|
||||
let rx = client
|
||||
.method_request(method.to_string(), params)
|
||||
.map_err(|_| RpcError::ClientError(Box::new(LightClientError::BackgroundClosed)))?;
|
||||
|
||||
// Fails if the background is closed.
|
||||
let response = rx
|
||||
.await
|
||||
.map_err(|_| RpcError::ClientError(Box::new(LightClientError::BackgroundClosed)))?;
|
||||
|
||||
tracing::trace!(target: LOG_TARGET, "RPC response {:?}", response);
|
||||
|
||||
response.map_err(|err| RpcError::ClientError(Box::new(err)))
|
||||
})
|
||||
}
|
||||
|
||||
fn subscribe_raw<'a>(
|
||||
&'a self,
|
||||
sub: &'a str,
|
||||
params: Option<Box<RawValue>>,
|
||||
_unsub: &'a str,
|
||||
) -> RpcFuture<'a, RpcSubscription> {
|
||||
let client = self.clone();
|
||||
|
||||
Box::pin(async move {
|
||||
tracing::trace!(
|
||||
target: LOG_TARGET,
|
||||
"Subscribe to {:?} with params {:?}",
|
||||
sub,
|
||||
params
|
||||
);
|
||||
|
||||
let params = match params {
|
||||
Some(params) => serde_json::to_string(¶ms).map_err(|_| {
|
||||
RpcError::ClientError(Box::new(LightClientError::InvalidParams))
|
||||
})?,
|
||||
None => "[]".into(),
|
||||
};
|
||||
|
||||
// Fails if the background is closed.
|
||||
let (sub_id, notif) = client
|
||||
.subscription_request(sub.to_string(), params)
|
||||
.map_err(|_| RpcError::ClientError(Box::new(LightClientError::BackgroundClosed)))?;
|
||||
|
||||
// Fails if the background is closed.
|
||||
let result = sub_id
|
||||
.await
|
||||
.map_err(|_| RpcError::ClientError(Box::new(LightClientError::BackgroundClosed)))?
|
||||
.map_err(|err| {
|
||||
RpcError::ClientError(Box::new(LightClientError::Request(err.to_string())))
|
||||
})?;
|
||||
|
||||
let sub_id = result
|
||||
.get()
|
||||
.trim_start_matches('"')
|
||||
.trim_end_matches('"')
|
||||
.to_string();
|
||||
tracing::trace!(target: LOG_TARGET, "Received subscription ID: {}", sub_id);
|
||||
|
||||
let stream = UnboundedReceiverStream::new(notif);
|
||||
|
||||
let rpc_substription_stream: Pin<
|
||||
Box<dyn Stream<Item = Result<Box<RawValue>, RpcError>> + Send + 'static>,
|
||||
> = Box::pin(stream.map(Ok));
|
||||
|
||||
let rpc_subscription: RpcSubscription = RpcSubscription {
|
||||
stream: rpc_substription_stream,
|
||||
id: Some(sub_id),
|
||||
};
|
||||
|
||||
Ok(rpc_subscription)
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -11,6 +11,9 @@
|
||||
mod offline_client;
|
||||
mod online_client;
|
||||
|
||||
#[cfg(feature = "unstable-light-client")]
|
||||
mod lightclient;
|
||||
|
||||
pub use offline_client::{OfflineClient, OfflineClientT};
|
||||
pub use online_client::{
|
||||
ClientRuntimeUpdater, OnlineClient, OnlineClientT, RuntimeUpdaterStream, Update, UpgradeError,
|
||||
@@ -18,3 +21,6 @@ pub use online_client::{
|
||||
|
||||
#[cfg(feature = "jsonrpsee")]
|
||||
pub use online_client::default_rpc_client;
|
||||
|
||||
#[cfg(feature = "unstable-light-client")]
|
||||
pub use lightclient::{LightClient, LightClientBuilder, LightClientError};
|
||||
@@ -8,6 +8,9 @@ mod dispatch_error;
|
||||
|
||||
use core::fmt::Debug;
|
||||
|
||||
#[cfg(feature = "unstable-light-client")]
|
||||
pub use crate::client::LightClientError;
|
||||
|
||||
// Re-export dispatch error types:
|
||||
pub use dispatch_error::{
|
||||
ArithmeticError, DispatchError, ModuleError, RawModuleError, TokenError, TransactionalError,
|
||||
@@ -64,6 +67,10 @@ pub enum Error {
|
||||
/// The bytes representing an error that we were unable to decode.
|
||||
#[error("An error occurred but it could not be decoded: {0:?}")]
|
||||
Unknown(Vec<u8>),
|
||||
/// Light client error.
|
||||
#[cfg(feature = "unstable-light-client")]
|
||||
#[error("An error occurred but it could not be decoded: {0:?}")]
|
||||
LightClient(#[from] LightClientError),
|
||||
/// Other error.
|
||||
#[error("Other error: {0}")]
|
||||
Other(String),
|
||||
|
||||
@@ -53,6 +53,11 @@ mod only_used_in_docs_or_tests {
|
||||
use tokio as _;
|
||||
}
|
||||
|
||||
// Suppress an unused dependency warning because tracing_subscriber is
|
||||
// only used in example code snippets at the time of writing.
|
||||
#[cfg(test)]
|
||||
use tracing_subscriber as _;
|
||||
|
||||
// Used to enable the js feature for wasm.
|
||||
#[cfg(feature = "web")]
|
||||
#[allow(unused_imports)]
|
||||
|
||||
@@ -239,6 +239,8 @@ pub type SystemProperties = serde_json::Map<String, serde_json::Value>;
|
||||
///
|
||||
/// This is copied from `sp-transaction-pool` to avoid a dependency on that crate. Therefore it
|
||||
/// must be kept compatible with that type from the target substrate version.
|
||||
///
|
||||
/// Substrate produces `camelCase` events, while smoldot produces `CamelCase` events.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub enum SubstrateTxStatus<Hash, BlockHash> {
|
||||
|
||||
@@ -13,7 +13,10 @@ homepage.workspace = true
|
||||
description = "Subxt integration tests that rely on the Substrate binary"
|
||||
|
||||
[features]
|
||||
default = ["subxt/integration-tests"]
|
||||
default = []
|
||||
|
||||
# Enable to run the tests with Light Client support.
|
||||
unstable-light-client = ["subxt/unstable-light-client"]
|
||||
|
||||
[dev-dependencies]
|
||||
assert_matches = { workspace = true }
|
||||
@@ -24,7 +27,6 @@ hex = { workspace = true }
|
||||
regex = { workspace = true }
|
||||
scale-info = { workspace = true, features = ["bit-vec"] }
|
||||
sp-core = { workspace = true }
|
||||
sp-runtime = { workspace = true }
|
||||
syn = { workspace = true }
|
||||
subxt = { workspace = true, features = ["unstable-metadata", "native", "jsonrpsee", "substrate-compat"] }
|
||||
subxt-signer = { workspace = true, features = ["subxt"] }
|
||||
@@ -37,3 +39,4 @@ tracing-subscriber = { workspace = true }
|
||||
wabt = { workspace = true }
|
||||
which = { workspace = true }
|
||||
substrate-runner = { workspace = true }
|
||||
sp-runtime = { workspace = true }
|
||||
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
@@ -0,0 +1,16 @@
|
||||
// Copyright 2019-2023 Parity Technologies (UK) Ltd.
|
||||
// This file is dual-licensed as Apache-2.0 or GPL-3.0.
|
||||
// see LICENSE for license details.
|
||||
|
||||
#[cfg(test)]
|
||||
mod blocks;
|
||||
#[cfg(test)]
|
||||
mod client;
|
||||
#[cfg(test)]
|
||||
mod frame;
|
||||
#[cfg(test)]
|
||||
mod metadata;
|
||||
#[cfg(test)]
|
||||
mod runtime_api;
|
||||
#[cfg(test)]
|
||||
mod storage;
|
||||
File renamed without changes.
File renamed without changes.
@@ -5,27 +5,28 @@
|
||||
#![deny(unused_crate_dependencies)]
|
||||
|
||||
#[cfg(test)]
|
||||
mod codegen;
|
||||
#[cfg(test)]
|
||||
mod utils;
|
||||
pub mod utils;
|
||||
|
||||
#[cfg(test)]
|
||||
mod blocks;
|
||||
#[cfg(test)]
|
||||
mod client;
|
||||
#[cfg(test)]
|
||||
mod frame;
|
||||
#[cfg(test)]
|
||||
mod metadata;
|
||||
#[cfg(test)]
|
||||
mod runtime_api;
|
||||
#[cfg(test)]
|
||||
mod storage;
|
||||
#[cfg_attr(test, allow(unused_imports))]
|
||||
use utils::*;
|
||||
|
||||
#[cfg(all(test, not(feature = "unstable-light-client")))]
|
||||
mod full_client;
|
||||
|
||||
#[cfg(all(test, feature = "unstable-light-client"))]
|
||||
mod light_client;
|
||||
|
||||
#[cfg(test)]
|
||||
use test_runtime::node_runtime;
|
||||
#[cfg(test)]
|
||||
use utils::*;
|
||||
|
||||
// These dependencies are used for the full client.
|
||||
#[cfg(all(test, not(feature = "unstable-light-client")))]
|
||||
use regex as _;
|
||||
#[cfg(all(test, not(feature = "unstable-light-client")))]
|
||||
use subxt_codegen as _;
|
||||
#[cfg(all(test, not(feature = "unstable-light-client")))]
|
||||
use syn as _;
|
||||
|
||||
// We don't use this dependency, but it's here so that we
|
||||
// can enable logging easily if need be. Add this to a test
|
||||
|
||||
@@ -0,0 +1,197 @@
|
||||
// Copyright 2019-2023 Parity Technologies (UK) Ltd.
|
||||
// This file is dual-licensed as Apache-2.0 or GPL-3.0.
|
||||
// see LICENSE for license details.
|
||||
|
||||
//! # Light Client Initialization and Testing
|
||||
//!
|
||||
//! The initialization process of the light client can be slow, especially when
|
||||
//! it needs to synchronize with a local running node for each individual
|
||||
//! #[tokio::test] in subxt. To optimize this process, a subset of tests is
|
||||
//! exposed to ensure the light client remains functional over time. Currently,
|
||||
//! these tests are placed under an unstable feature flag.
|
||||
//!
|
||||
//! Ideally, we would place the light client initialization in a shared static
|
||||
//! using `OnceCell`. However, during the initialization, tokio::spawn is used
|
||||
//! to multiplex between subxt requests and node responses. The #[tokio::test]
|
||||
//! macro internally creates a new Runtime for each individual test. This means
|
||||
//! that only the first test, which spawns the substrate binary and synchronizes
|
||||
//! the light client, would have access to the background task. The cleanup process
|
||||
//! would destroy the spawned background task, preventing subsequent tests from
|
||||
//! accessing it.
|
||||
//!
|
||||
//! To address this issue, we can consider creating a slim proc-macro that
|
||||
//! transforms the #[tokio::test] into a plain #[test] and runs all the tests
|
||||
//! on a shared tokio runtime. This approach would allow multiple tests to share
|
||||
//! the same background task, ensuring consistent access to the light client.
|
||||
//!
|
||||
//! For more context see: https://github.com/tokio-rs/tokio/issues/2374.
|
||||
//!
|
||||
|
||||
use crate::utils::node_runtime;
|
||||
use codec::{Compact, Encode};
|
||||
use futures::StreamExt;
|
||||
use subxt::{
|
||||
client::{LightClient, LightClientBuilder, OfflineClientT, OnlineClientT},
|
||||
config::PolkadotConfig,
|
||||
rpc::types::FollowEvent,
|
||||
};
|
||||
use subxt_metadata::Metadata;
|
||||
|
||||
// We don't use these dependencies.
|
||||
use assert_matches as _;
|
||||
use frame_metadata as _;
|
||||
use hex as _;
|
||||
use regex as _;
|
||||
use scale_info as _;
|
||||
use sp_core as _;
|
||||
use sp_runtime as _;
|
||||
use subxt_codegen as _;
|
||||
use subxt_signer as _;
|
||||
use syn as _;
|
||||
use tracing as _;
|
||||
use wabt as _;
|
||||
|
||||
type Client = LightClient<PolkadotConfig>;
|
||||
|
||||
// Check that we can subscribe to non-finalized blocks.
|
||||
async fn non_finalized_headers_subscription(api: &Client) -> Result<(), subxt::Error> {
|
||||
let mut sub = api.blocks().subscribe_best().await?;
|
||||
let header = sub.next().await.unwrap()?;
|
||||
let block_hash = header.hash();
|
||||
let current_block_hash = api.rpc().block_hash(None).await?.unwrap();
|
||||
|
||||
assert_eq!(block_hash, current_block_hash);
|
||||
|
||||
let _block = sub.next().await.unwrap()?;
|
||||
let _block = sub.next().await.unwrap()?;
|
||||
let _block = sub.next().await.unwrap()?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// Check that we can subscribe to finalized blocks.
|
||||
async fn finalized_headers_subscription(api: &Client) -> Result<(), subxt::Error> {
|
||||
let mut sub = api.blocks().subscribe_finalized().await?;
|
||||
let header = sub.next().await.unwrap()?;
|
||||
let finalized_hash = api.rpc().finalized_head().await?;
|
||||
|
||||
assert_eq!(header.hash(), finalized_hash);
|
||||
|
||||
let _block = sub.next().await.unwrap()?;
|
||||
let _block = sub.next().await.unwrap()?;
|
||||
let _block = sub.next().await.unwrap()?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// Check that we can subscribe to non-finalized blocks.
|
||||
async fn runtime_api_call(api: &Client) -> Result<(), subxt::Error> {
|
||||
let mut sub = api.blocks().subscribe_best().await?;
|
||||
|
||||
let block = sub.next().await.unwrap()?;
|
||||
let rt = block.runtime_api().await?;
|
||||
|
||||
// get metadata via state_call.
|
||||
let (_, meta1) = rt
|
||||
.call_raw::<(Compact<u32>, Metadata)>("Metadata_metadata", None)
|
||||
.await?;
|
||||
|
||||
// get metadata via `state_getMetadata`.
|
||||
let meta2 = api.rpc().metadata_legacy(None).await?;
|
||||
|
||||
// They should be the same.
|
||||
assert_eq!(meta1.encode(), meta2.encode());
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// Lookup for the `Timestamp::now` plain storage entry.
|
||||
async fn storage_plain_lookup(api: &Client) -> Result<(), subxt::Error> {
|
||||
let addr = node_runtime::storage().timestamp().now();
|
||||
let entry = api
|
||||
.storage()
|
||||
.at_latest()
|
||||
.await?
|
||||
.fetch_or_default(&addr)
|
||||
.await?;
|
||||
assert!(entry > 0);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// Subscribe to produced blocks using the `ChainHead` spec V2 and fetch the header of
|
||||
// just a few reported blocks.
|
||||
async fn follow_chain_head(api: &Client) -> Result<(), subxt::Error> {
|
||||
let mut blocks = api.rpc().chainhead_unstable_follow(false).await?;
|
||||
let sub_id = blocks
|
||||
.subscription_id()
|
||||
.expect("RPC provides a valid subscription id; qed")
|
||||
.to_owned();
|
||||
|
||||
let event = blocks.next().await.unwrap()?;
|
||||
if let FollowEvent::BestBlockChanged(best_block) = event {
|
||||
let hash = best_block.best_block_hash;
|
||||
let _header = api
|
||||
.rpc()
|
||||
.chainhead_unstable_header(sub_id.clone(), hash)
|
||||
.await?
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
let event = blocks.next().await.unwrap()?;
|
||||
if let FollowEvent::BestBlockChanged(best_block) = event {
|
||||
let hash = best_block.best_block_hash;
|
||||
let _header = api
|
||||
.rpc()
|
||||
.chainhead_unstable_header(sub_id.clone(), hash)
|
||||
.await?
|
||||
.unwrap();
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// Make a dynamic constant query for `System::BlockLenght`.
|
||||
async fn dynamic_constant_query(api: &Client) -> Result<(), subxt::Error> {
|
||||
let constant_query = subxt::dynamic::constant("System", "BlockLength");
|
||||
let _value = api.constants().at(&constant_query)?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// Fetch a few all events from the latest block and decode them dynamically.
|
||||
async fn dynamic_events(api: &Client) -> Result<(), subxt::Error> {
|
||||
let events = api.events().at_latest().await?;
|
||||
|
||||
for event in events.iter() {
|
||||
let _event = event?;
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// Make a few raw RPC calls to the chain.
|
||||
async fn various_rpc_calls(api: &Client) -> Result<(), subxt::Error> {
|
||||
let _system_chain = api.rpc().system_chain().await?;
|
||||
let _system_name = api.rpc().system_name().await?;
|
||||
let _finalized_hash = api.rpc().finalized_head().await?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn light_client_testing() -> Result<(), subxt::Error> {
|
||||
let api: LightClient<PolkadotConfig> = LightClientBuilder::new()
|
||||
.build_from_url("wss://rpc.polkadot.io:443")
|
||||
.await?;
|
||||
|
||||
non_finalized_headers_subscription(&api).await?;
|
||||
finalized_headers_subscription(&api).await?;
|
||||
runtime_api_call(&api).await?;
|
||||
storage_plain_lookup(&api).await?;
|
||||
follow_chain_head(&api).await?;
|
||||
dynamic_constant_query(&api).await?;
|
||||
dynamic_events(&api).await?;
|
||||
various_rpc_calls(&api).await?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
@@ -2,7 +2,7 @@
|
||||
// This file is dual-licensed as Apache-2.0 or GPL-3.0.
|
||||
// see LICENSE for license details.
|
||||
|
||||
pub(crate) use crate::{node_runtime, TestNodeProcess};
|
||||
pub(crate) use crate::{node_runtime, utils::TestNodeProcess};
|
||||
|
||||
use subxt::SubstrateConfig;
|
||||
|
||||
|
||||
@@ -6,11 +6,19 @@ use std::ffi::{OsStr, OsString};
|
||||
use substrate_runner::SubstrateNode;
|
||||
use subxt::{Config, OnlineClient};
|
||||
|
||||
#[cfg(feature = "unstable-light-client")]
|
||||
use subxt::client::{LightClient, LightClientBuilder};
|
||||
|
||||
/// Spawn a local substrate node for testing subxt.
|
||||
pub struct TestNodeProcess<R: Config> {
|
||||
// Keep a handle to the node; once it's dropped the node is killed.
|
||||
_proc: SubstrateNode,
|
||||
|
||||
#[cfg(not(feature = "unstable-light-client"))]
|
||||
client: OnlineClient<R>,
|
||||
|
||||
#[cfg(feature = "unstable-light-client")]
|
||||
client: LightClient<R>,
|
||||
}
|
||||
|
||||
impl<R> TestNodeProcess<R>
|
||||
@@ -26,9 +34,16 @@ where
|
||||
}
|
||||
|
||||
/// Returns the subxt client connected to the running node.
|
||||
#[cfg(not(feature = "unstable-light-client"))]
|
||||
pub fn client(&self) -> OnlineClient<R> {
|
||||
self.client.clone()
|
||||
}
|
||||
|
||||
/// Returns the subxt client connected to the running node.
|
||||
#[cfg(feature = "unstable-light-client")]
|
||||
pub fn client(&self) -> LightClient<R> {
|
||||
self.client.clone()
|
||||
}
|
||||
}
|
||||
|
||||
/// Construct a test node process.
|
||||
@@ -71,8 +86,13 @@ impl TestNodeProcessBuilder {
|
||||
let proc = node_builder.spawn().map_err(|e| e.to_string())?;
|
||||
let ws_url = format!("ws://127.0.0.1:{}", proc.ws_port());
|
||||
|
||||
#[cfg(feature = "unstable-light-client")]
|
||||
let client = build_light_client(&proc).await;
|
||||
|
||||
// Connect to the node with a subxt client:
|
||||
#[cfg(not(feature = "unstable-light-client"))]
|
||||
let client = OnlineClient::from_url(ws_url.clone()).await;
|
||||
|
||||
match client {
|
||||
Ok(client) => Ok(TestNodeProcess {
|
||||
_proc: proc,
|
||||
@@ -82,3 +102,30 @@ impl TestNodeProcessBuilder {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(feature = "unstable-light-client")]
|
||||
async fn build_light_client<R: Config>(proc: &SubstrateNode) -> Result<LightClient<R>, String> {
|
||||
// RPC endpoint.
|
||||
let ws_url = format!("ws://127.0.0.1:{}", proc.ws_port());
|
||||
|
||||
// Step 1. Wait for a few blocks to be produced using the subxt client.
|
||||
let client = OnlineClient::<R>::from_url(ws_url.clone())
|
||||
.await
|
||||
.map_err(|err| format!("Failed to connect to node rpc at {ws_url}: {err}"))?;
|
||||
|
||||
super::wait_for_blocks(&client).await;
|
||||
|
||||
// Step 2. Construct the light client.
|
||||
// P2p bootnode.
|
||||
let bootnode = format!(
|
||||
"/ip4/127.0.0.1/tcp/{}/p2p/{}",
|
||||
proc.p2p_port(),
|
||||
proc.p2p_address()
|
||||
);
|
||||
|
||||
LightClientBuilder::new()
|
||||
.bootnodes([bootnode.as_str()])
|
||||
.build_from_url(ws_url.as_str())
|
||||
.await
|
||||
.map_err(|e| format!("Failed to construct light client {}", e.to_string()))
|
||||
}
|
||||
@@ -7,8 +7,20 @@ use subxt::{client::OnlineClientT, Config};
|
||||
/// Wait for blocks to be produced before running tests. Waiting for two blocks
|
||||
/// (the genesis block and another one) seems to be enough to allow tests
|
||||
/// like `dry_run_passes` to work properly.
|
||||
///
|
||||
/// If the "unstable-light-client" feature flag is enabled, this will wait for
|
||||
/// 5 blocks instead of two. The light client needs the extra blocks to avoid
|
||||
/// errors caused by loading information that is not available in the first 2 blocks
|
||||
/// (`Failed to load the block weight for block`).
|
||||
pub async fn wait_for_blocks<C: Config>(api: &impl OnlineClientT<C>) {
|
||||
let mut sub = api.rpc().subscribe_all_block_headers().await.unwrap();
|
||||
sub.next().await;
|
||||
sub.next().await;
|
||||
|
||||
#[cfg(feature = "unstable-light-client")]
|
||||
{
|
||||
sub.next().await;
|
||||
sub.next().await;
|
||||
sub.next().await;
|
||||
}
|
||||
}
|
||||
@@ -6,6 +6,8 @@
|
||||
pub enum Error {
|
||||
Io(std::io::Error),
|
||||
CouldNotExtractPort,
|
||||
CouldNotExtractP2pAddress,
|
||||
CouldNotExtractP2pPort,
|
||||
}
|
||||
|
||||
impl std::fmt::Display for Error {
|
||||
@@ -16,6 +18,14 @@ impl std::fmt::Display for Error {
|
||||
f,
|
||||
"could not extract port from running substrate node's stdout"
|
||||
),
|
||||
Error::CouldNotExtractP2pAddress => write!(
|
||||
f,
|
||||
"could not extract p2p address from running substrate node's stdout"
|
||||
),
|
||||
Error::CouldNotExtractP2pPort => write!(
|
||||
f,
|
||||
"could not extract p2p port from running substrate node's stdout"
|
||||
),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -56,7 +56,7 @@ impl SubstrateNodeBuilder {
|
||||
pub fn spawn(self) -> Result<SubstrateNode, Error> {
|
||||
let mut cmd = Command::new(self.binary_path);
|
||||
|
||||
cmd.env("RUST_LOG", "info")
|
||||
cmd.env("RUST_LOG", "info,libp2p_tcp=debug")
|
||||
.stdout(process::Stdio::piped())
|
||||
.stderr(process::Stdio::piped())
|
||||
.arg("--dev")
|
||||
@@ -74,16 +74,26 @@ impl SubstrateNodeBuilder {
|
||||
|
||||
// Wait for RPC port to be logged (it's logged to stderr).
|
||||
let stderr = proc.stderr.take().unwrap();
|
||||
let ws_port =
|
||||
try_find_substrate_port_from_output(stderr).ok_or(Error::CouldNotExtractPort)?;
|
||||
let (ws_port, p2p_address, p2p_port) = try_find_substrate_port_from_output(stderr);
|
||||
|
||||
Ok(SubstrateNode { proc, ws_port })
|
||||
let ws_port = ws_port.ok_or(Error::CouldNotExtractPort)?;
|
||||
let p2p_address = p2p_address.ok_or(Error::CouldNotExtractP2pAddress)?;
|
||||
let p2p_port = p2p_port.ok_or(Error::CouldNotExtractP2pPort)?;
|
||||
|
||||
Ok(SubstrateNode {
|
||||
proc,
|
||||
ws_port,
|
||||
p2p_address,
|
||||
p2p_port,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
pub struct SubstrateNode {
|
||||
proc: process::Child,
|
||||
ws_port: u16,
|
||||
p2p_address: String,
|
||||
p2p_port: u32,
|
||||
}
|
||||
|
||||
impl SubstrateNode {
|
||||
@@ -102,6 +112,16 @@ impl SubstrateNode {
|
||||
self.ws_port
|
||||
}
|
||||
|
||||
/// Return the libp2p address of the running node.
|
||||
pub fn p2p_address(&self) -> String {
|
||||
self.p2p_address.clone()
|
||||
}
|
||||
|
||||
/// Return the libp2p port of the running node.
|
||||
pub fn p2p_port(&self) -> u32 {
|
||||
self.p2p_port
|
||||
}
|
||||
|
||||
/// Kill the process.
|
||||
pub fn kill(&mut self) -> std::io::Result<()> {
|
||||
self.proc.kill()
|
||||
@@ -116,28 +136,63 @@ impl Drop for SubstrateNode {
|
||||
|
||||
// Consume a stderr reader from a spawned substrate command and
|
||||
// locate the port number that is logged out to it.
|
||||
fn try_find_substrate_port_from_output(r: impl Read + Send + 'static) -> Option<u16> {
|
||||
BufReader::new(r).lines().take(50).find_map(|line| {
|
||||
fn try_find_substrate_port_from_output(
|
||||
r: impl Read + Send + 'static,
|
||||
) -> (Option<u16>, Option<String>, Option<u32>) {
|
||||
let mut port = None;
|
||||
let mut p2p_address = None;
|
||||
let mut p2p_port = None;
|
||||
|
||||
for line in BufReader::new(r).lines().take(50) {
|
||||
let line = line.expect("failed to obtain next line from stdout for port discovery");
|
||||
|
||||
// does the line contain our port (we expect this specific output from substrate).
|
||||
let line_end = line
|
||||
// Parse the port lines
|
||||
let line_port = line
|
||||
// oldest message:
|
||||
.rsplit_once("Listening for new connections on 127.0.0.1:")
|
||||
// slightly newer message:
|
||||
.or_else(|| line.rsplit_once("Running JSON-RPC WS server: addr=127.0.0.1:"))
|
||||
// newest message (jsonrpsee merging http and ws servers):
|
||||
.or_else(|| line.rsplit_once("Running JSON-RPC server: addr=127.0.0.1:"))
|
||||
.map(|(_, port_str)| port_str)?;
|
||||
.map(|(_, port_str)| port_str);
|
||||
|
||||
// trim non-numeric chars from the end of the port part of the line.
|
||||
let port_str = line_end.trim_end_matches(|b: char| !b.is_ascii_digit());
|
||||
if let Some(line_port) = line_port {
|
||||
// trim non-numeric chars from the end of the port part of the line.
|
||||
let port_str = line_port.trim_end_matches(|b: char| !b.is_ascii_digit());
|
||||
|
||||
// expect to have a number here (the chars after '127.0.0.1:') and parse them into a u16.
|
||||
let port_num = port_str
|
||||
.parse()
|
||||
.unwrap_or_else(|_| panic!("valid port expected for log line, got '{port_str}'"));
|
||||
// expect to have a number here (the chars after '127.0.0.1:') and parse them into a u16.
|
||||
let port_num = port_str
|
||||
.parse()
|
||||
.unwrap_or_else(|_| panic!("valid port expected for log line, got '{port_str}'"));
|
||||
port = Some(port_num);
|
||||
}
|
||||
|
||||
Some(port_num)
|
||||
})
|
||||
// Parse the p2p address line
|
||||
let line_address = line
|
||||
.rsplit_once("Local node identity is: ")
|
||||
.map(|(_, address_str)| address_str);
|
||||
|
||||
if let Some(line_address) = line_address {
|
||||
let address = line_address.trim_end_matches(|b: char| b.is_ascii_whitespace());
|
||||
p2p_address = Some(address.into());
|
||||
}
|
||||
|
||||
// Parse the p2p port line (present in debug logs)
|
||||
let p2p_port_line = line
|
||||
.rsplit_once("libp2p_tcp: New listen address: /ip4/127.0.0.1/tcp/")
|
||||
.map(|(_, address_str)| address_str);
|
||||
|
||||
if let Some(line_port) = p2p_port_line {
|
||||
// trim non-numeric chars from the end of the port part of the line.
|
||||
let port_str = line_port.trim_end_matches(|b: char| !b.is_ascii_digit());
|
||||
|
||||
// expect to have a number here (the chars after '127.0.0.1:') and parse them into a u16.
|
||||
let port_num = port_str
|
||||
.parse()
|
||||
.unwrap_or_else(|_| panic!("valid port expected for log line, got '{port_str}'"));
|
||||
p2p_port = Some(port_num);
|
||||
}
|
||||
}
|
||||
|
||||
(port, p2p_address, p2p_port)
|
||||
}
|
||||
Reference in new issue
Block a user