mirror of
https://github.com/pezkuwichain/pezkuwi-subxt.git
synced 2026-05-31 00:31:02 +00:00
Update bridges subtree (#5165)
* Squashed 'bridges/' changes from 1602249f0a..f220d2fcca f220d2fcca Polkadot staging update (#1356) 02fd3d497c fix parse_transaction on Rialto+Millau (#1360) bc191fd9a2 update parity-scale-codec to 3.1.2 (#1359) a37226e79c update chain versions (#1358) ff5d539fcb Update Substrate/Polkadot/Cumulus references (#1353) 1581f60cd5 Support dedicated lanes for pallets (#962) 0a7ccf5c57 ignore more "increase" alerts that are sometimes signalling NoData at startup (#1351) 31165127cc added no_stack_overflow_when_decoding_nested_call_during_dispatch test (#1349) 7000619eb8 replace From<>InboundLaneApi with direct storage reads (#1348) 515df10ccc added alerts for relay balances (#1347) b56f6a87de Mortal conversion rate updater transactions (#1257) 20f2f331ec edition = "2021" (#1346) 99147d4f75 update regex to 1.5.5 (#1345) 686191f379 use DecodeLimit when decoding incoming calls (#1344) a70c276006 get rid of '[No Data] Messages from Millau to Rialto are not being delivered' warnings (#1342) 01f29b8ac1 fix conversion rate metric in dashboards (#1341) 51c3bf351f Increase rate from metric when estimating fee (#1340) 3bb9c4f68f fix generator scripts to be consistent with updatedrelay output (#1339) 0475a1667b fixed mess with conversion rates (#1338) d8fdd7d716 synchronize relay cli changes and token swap generator script (#1337) 6e928137a5 fix conversion rate override in token swap (#1336) 62d4a4811d override conversion rate in tokens swap generator (#1335) ed9e1c839c fi typo in generator script (#1334) 3254b5af7a Override conversion rate when computing message fee (#1261) 66df68b5b8 Revert "Revert "override conversion rate in estimate-message-fee RPC (#1189)" (#1275)" (#1333) 0ca6fc6ef8 fix clippy issues (#1332) 5414b2fffb Reinitialize bridge relay subcommand (#1331) a63d95ba7d removed extra *_RUNTIME_VERSION consts from relay code (#1330) 59fb18a310 fix typo in alert expression (#1329) a6267a47ee Using-same-fork metric for finality and complex relay (#1327) 88d684d37e use mortal transactions in transaction resubmitter (#1326) 8ff88b6844 impl Decode for SignedExtensions (otherwise transaction resubmitter panicks) (#1325) 1ed09854f0 Encode and estimate Rococo/Wococo/Kusama/Polkadot messages (#1322) ddb4517e13 Add some tests to check integrity of chain constants + bridge configuration (#1316) bdeedb7ab9 Fix issues from cargo deny (#1311) d3d79d01e0 expose fee multiplier metrics in messages relay (#1312) c8b3f0ea16 Endow relayer account at target chain in message benchmarks (#1310) f51ecd92b6 fix benchmarks before using it in Polkadot/Kusama/Rococo runtimes (#1309) 6935c619ad increase relay balance guard limits for Polkadot<>Kusama bridge (#1308) 7e31834c66 Fix mandatory headers scanning in on-demand relay (#1306) 92ddc3ea7a Polkadot-staging update (#1305) 3787193a31 fix session length of Rococo and Wococo (#1304) eb468d29c0 Revert nightly docker pin (#1301) e2d4c073e1 Use raw balance value if tokenDecimals property is missing (#1299) 108f4b29d1 Fix ss58 prefixes of Polkadot, Kusama and Westend used by relay (#1298) 64fbd2705e bump chain spec versions (#1297) 5707777b86 Bump Substrate/Polkadot/Cumulus refs (#1295) 29eecdf1fa Merge pull request #1294 from paritytech/polkadot-staging-update 1f0c05368e Relay balance metrics (#1291) 6356bb90b3 when messages pallet is halted, relay shall not submit messages delivery/confirmation transactions (#1289) 800dc2df8d when GRANDPA pallet is halted, relay shall not submit finality transactions (#1288) 3dd8e4f936 disable BEEFY allerts for Rialto (#1285) f58fed7380 support version mode cli options in send-message subcommand (#1284) 3aac448da3 reuse polkadot-service code (#1273) 2bdbb651e1 replace latest_confirmed_nonce runtime APIs with direct storage reads (#1282) 5f9c6d241f move "common" code of messages pallet benchmarks helpers to the common library (#1281) 173d2d8229 Merge pull request #1280 from paritytech/polkadot-staging-update 8b9c4ec16d do not start spec_version guard when version mode is set to auto (#1278) e98d682de2 removed extra messages benchmarks (#1279) c730e25b61 Move benchmarks from Rialto to Millau (#1277) 54146416e7 Merge pull request #1276 from paritytech/polkadot-staging-update df70118174 Merge branch 'master' into polkadot-staging-update ed7def64c4 Revert "override conversion rate in estimate-message-fee RPC (#1189)" (#1275) 38c6c3a49f Use "production" floating tag when uilding docker image from version git tags (#1272) ded9ff6dbb Replace InboundLaneApi::latest_received_nonce with direct storage read (#1269) f704a741ee Polkadot staging update (#1270) 8c65f0d7ab verify that GRANDPA pallet is not initialized before submitting initialization transaction (#1267) e7e83d8944 remove OutboundLaneApi::latest_received_nonce (#1262) 9f4b34acf1 bump rococo version (#1263) 82c08c5a87 read latest_generated_nonce directly from storage (#1260) 50ffb5dd08 override conversion rate in estimate-message-fee RPC (#1189) 467ca5ef59 move storage keys computation to primitivs (#1254) 4f9884066b remporary use pinned bridges-ci image in Dockerfile (#1258) edfcb74e00 Change submit transaction spec_version and transaction_version query from chain (#1248) 4009d970d0 pin bridges-ci image (#1256) 65e51b5e1c decrease startup sleep to 5s for relays and to 120s for generators + remove curl (#1251) 3bc74355d9 Add missing RPC APIs to rialto parachain node (#1250) 80c9429284 Bump relay version to 1.0.0 (#1249) 9ead06af2a runtimes: fix call_size() test (#1245) 4fc8a29357 Use same endowed accounts set on dev/local chains (#1244) fed54371c2 Refactor message relay helpers (#1234) a15b4faae7 post-merge build fix (#1243) 52232d8d54 Fix transactions mortality (#1196) c07bba931f Expose prometheus BEEFY metrics and add them to grafana dashboard (#1242) f927775bd5 Refactor finality relay helpers (#1220) 7bf76f14a8 Update Rococo/Wococo version + prepare relay for Rococo<>Wococo bridge (#1241) e860fecd04 Enable offchain indexing for Rialto/Millau nodes (#1239) 04d4d1c6b4 Enable Beefy debug logs in test deployment (#1237) cd771f1089 Fix storage parameter name computation (#1238) 816ddd2dd2 Integrate BEEFY with Rialto & Millau runtimes (#1227) d94b62b1ac update dependencies (#1229) 98eb9ee13d Add mut support (#1232) ffef6f89f9 fixed set_operational in GRANDPA pallet (#1226) bd2f8bfbd7 Add CODEOWNERS file (#1219) 6b5cf2b591 Unify metric names (#1209) d1541e797e remove abandoned exchange relay (#1217) 39140d0b34 Remove unused `relays/headers` (#1216) 9bc071d42b Remove unused PoA<>Substrate bridge (#1210) 877e8d01e3 Fix UI deployment. (#1211) 6cd5775ebe Add `AtLeast32BitUnsigned` for MessageLance::SourceChainBalance (#1207) git-subtree-dir: bridges git-subtree-split: f220d2fccabbf141101d19456ecb4e3576a1d797 * fix compilation warnings
This commit is contained in:
Svyatoslav Nikolsky
committed by
GitHub
GitHub
parent
20da356434
commit
8e01ba9c03
@@ -2,7 +2,7 @@
|
||||
name = "finality-relay"
|
||||
version = "0.1.0"
|
||||
authors = ["Parity Technologies <admin@parity.io>"]
|
||||
edition = "2018"
|
||||
edition = "2021"
|
||||
license = "GPL-3.0-or-later WITH Classpath-exception-2.0"
|
||||
description = "Finality proofs relay"
|
||||
|
||||
|
||||
@@ -29,7 +29,7 @@ use futures::{select, Future, FutureExt, Stream, StreamExt};
|
||||
use num_traits::{One, Saturating};
|
||||
use relay_utils::{
|
||||
metrics::MetricsParams, relay_loop::Client as RelayClient, retry_backoff, FailedClient,
|
||||
MaybeConnectionError,
|
||||
HeaderId, MaybeConnectionError,
|
||||
};
|
||||
use std::{
|
||||
pin::Pin,
|
||||
@@ -87,7 +87,9 @@ pub trait SourceClient<P: FinalitySyncPipeline>: RelayClient {
|
||||
#[async_trait]
|
||||
pub trait TargetClient<P: FinalitySyncPipeline>: RelayClient {
|
||||
/// Get best finalized source block number.
|
||||
async fn best_finalized_source_block_number(&self) -> Result<P::Number, Self::Error>;
|
||||
async fn best_finalized_source_block_id(
|
||||
&self,
|
||||
) -> Result<HeaderId<P::Hash, P::Number>, Self::Error>;
|
||||
|
||||
/// Submit header finality proof.
|
||||
async fn submit_finality_proof(
|
||||
@@ -114,7 +116,11 @@ pub async fn run<P: FinalitySyncPipeline>(
|
||||
let exit_signal = exit_signal.shared();
|
||||
relay_utils::relay_loop(source_client, target_client)
|
||||
.with_metrics(metrics_params)
|
||||
.loop_metric(SyncLoopMetrics::new(Some(&metrics_prefix::<P>()))?)?
|
||||
.loop_metric(SyncLoopMetrics::new(
|
||||
Some(&metrics_prefix::<P>()),
|
||||
"source",
|
||||
"source_at_target",
|
||||
)?)?
|
||||
.expose()
|
||||
.await?
|
||||
.run(metrics_prefix::<P>(), move |source_client, target_client, metrics| {
|
||||
@@ -169,7 +175,7 @@ where
|
||||
|
||||
/// Information about transaction that we have submitted.
|
||||
#[derive(Debug, Clone)]
|
||||
struct Transaction<Number> {
|
||||
pub(crate) struct Transaction<Number> {
|
||||
/// Time when we have submitted this transaction.
|
||||
pub time: Instant,
|
||||
/// The number of the header we have submitted.
|
||||
@@ -181,7 +187,7 @@ pub(crate) struct RestartableFinalityProofsStream<S> {
|
||||
/// Flag that the stream needs to be restarted.
|
||||
pub(crate) needs_restart: bool,
|
||||
/// The stream itself.
|
||||
stream: Pin<Box<S>>,
|
||||
pub(crate) stream: Pin<Box<S>>,
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
@@ -192,15 +198,16 @@ impl<S> From<S> for RestartableFinalityProofsStream<S> {
|
||||
}
|
||||
|
||||
/// Finality synchronization loop state.
|
||||
struct FinalityLoopState<'a, P: FinalitySyncPipeline, FinalityProofsStream> {
|
||||
pub(crate) struct FinalityLoopState<'a, P: FinalitySyncPipeline, FinalityProofsStream> {
|
||||
/// Synchronization loop progress.
|
||||
progress: &'a mut (Instant, Option<P::Number>),
|
||||
pub(crate) progress: &'a mut (Instant, Option<P::Number>),
|
||||
/// Finality proofs stream.
|
||||
finality_proofs_stream: &'a mut RestartableFinalityProofsStream<FinalityProofsStream>,
|
||||
pub(crate) finality_proofs_stream:
|
||||
&'a mut RestartableFinalityProofsStream<FinalityProofsStream>,
|
||||
/// Recent finality proofs that we have read from the stream.
|
||||
recent_finality_proofs: &'a mut FinalityProofs<P>,
|
||||
pub(crate) recent_finality_proofs: &'a mut FinalityProofs<P>,
|
||||
/// Last transaction that we have submitted to the target node.
|
||||
last_transaction: Option<Transaction<P::Number>>,
|
||||
pub(crate) last_transaction: Option<Transaction<P::Number>>,
|
||||
}
|
||||
|
||||
async fn run_until_connection_lost<P: FinalitySyncPipeline>(
|
||||
@@ -280,7 +287,7 @@ async fn run_until_connection_lost<P: FinalitySyncPipeline>(
|
||||
}
|
||||
}
|
||||
|
||||
async fn run_loop_iteration<P, SC, TC>(
|
||||
pub(crate) async fn run_loop_iteration<P, SC, TC>(
|
||||
source_client: &SC,
|
||||
target_client: &TC,
|
||||
state: FinalityLoopState<'_, P, SC::FinalityProofsStream>,
|
||||
@@ -295,13 +302,31 @@ where
|
||||
// read best source headers ids from source and target nodes
|
||||
let best_number_at_source =
|
||||
source_client.best_finalized_block_number().await.map_err(Error::Source)?;
|
||||
let best_number_at_target = target_client
|
||||
.best_finalized_source_block_number()
|
||||
let best_id_at_target =
|
||||
target_client.best_finalized_source_block_id().await.map_err(Error::Target)?;
|
||||
let best_number_at_target = best_id_at_target.0;
|
||||
|
||||
let different_hash_at_source = ensure_same_fork::<P, _>(&best_id_at_target, source_client)
|
||||
.await
|
||||
.map_err(Error::Target)?;
|
||||
.map_err(Error::Source)?;
|
||||
let using_same_fork = different_hash_at_source.is_none();
|
||||
if let Some(ref different_hash_at_source) = different_hash_at_source {
|
||||
log::error!(
|
||||
target: "bridge",
|
||||
"Source node ({}) and pallet at target node ({}) have different headers at the same height {:?}: \
|
||||
at-source {:?} vs at-target {:?}",
|
||||
P::SOURCE_NAME,
|
||||
P::TARGET_NAME,
|
||||
best_number_at_target,
|
||||
different_hash_at_source,
|
||||
best_id_at_target.1,
|
||||
);
|
||||
}
|
||||
|
||||
if let Some(ref metrics_sync) = *metrics_sync {
|
||||
metrics_sync.update_best_block_at_source(best_number_at_source);
|
||||
metrics_sync.update_best_block_at_target(best_number_at_target);
|
||||
metrics_sync.update_using_same_fork(using_same_fork);
|
||||
}
|
||||
*state.progress =
|
||||
print_sync_progress::<P>(*state.progress, best_number_at_source, best_number_at_target);
|
||||
@@ -427,6 +452,22 @@ where
|
||||
Ok(selected_finality_proof)
|
||||
}
|
||||
|
||||
/// Ensures that both clients are on the same fork.
|
||||
///
|
||||
/// Returns `Some(_)` with header has at the source client if headers are different.
|
||||
async fn ensure_same_fork<P: FinalitySyncPipeline, SC: SourceClient<P>>(
|
||||
best_id_at_target: &HeaderId<P::Hash, P::Number>,
|
||||
source_client: &SC,
|
||||
) -> Result<Option<P::Hash>, SC::Error> {
|
||||
let header_at_source = source_client.header_and_finality_proof(best_id_at_target.0).await?.0;
|
||||
let header_hash_at_source = header_at_source.hash();
|
||||
Ok(if best_id_at_target.1 == header_hash_at_source {
|
||||
None
|
||||
} else {
|
||||
Some(header_hash_at_source)
|
||||
})
|
||||
}
|
||||
|
||||
/// Finality proof that has been selected by the `read_missing_headers` function.
|
||||
pub(crate) enum SelectedFinalityProof<Header, FinalityProof> {
|
||||
/// Mandatory header and its proof has been selected. We shall submit proof for this header.
|
||||
|
||||
@@ -20,10 +20,12 @@
|
||||
|
||||
use crate::{
|
||||
finality_loop::{
|
||||
prune_recent_finality_proofs, read_finality_proofs_from_stream, run,
|
||||
select_better_recent_finality_proof, select_header_to_submit, FinalityProofs,
|
||||
FinalitySyncParams, RestartableFinalityProofsStream, SourceClient, TargetClient,
|
||||
prune_recent_finality_proofs, read_finality_proofs_from_stream, run, run_loop_iteration,
|
||||
select_better_recent_finality_proof, select_header_to_submit, FinalityLoopState,
|
||||
FinalityProofs, FinalitySyncParams, RestartableFinalityProofsStream, SourceClient,
|
||||
TargetClient,
|
||||
},
|
||||
sync_loop_metrics::SyncLoopMetrics,
|
||||
FinalityProof, FinalitySyncPipeline, SourceHeader,
|
||||
};
|
||||
|
||||
@@ -31,12 +33,18 @@ use async_trait::async_trait;
|
||||
use futures::{FutureExt, Stream, StreamExt};
|
||||
use parking_lot::Mutex;
|
||||
use relay_utils::{
|
||||
metrics::MetricsParams, relay_loop::Client as RelayClient, MaybeConnectionError,
|
||||
metrics::MetricsParams, relay_loop::Client as RelayClient, HeaderId, MaybeConnectionError,
|
||||
};
|
||||
use std::{
|
||||
collections::HashMap,
|
||||
pin::Pin,
|
||||
sync::Arc,
|
||||
time::{Duration, Instant},
|
||||
};
|
||||
use std::{collections::HashMap, pin::Pin, sync::Arc, time::Duration};
|
||||
|
||||
type IsMandatory = bool;
|
||||
type TestNumber = u64;
|
||||
type TestHash = u64;
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
enum TestError {
|
||||
@@ -56,16 +64,20 @@ impl FinalitySyncPipeline for TestFinalitySyncPipeline {
|
||||
const SOURCE_NAME: &'static str = "TestSource";
|
||||
const TARGET_NAME: &'static str = "TestTarget";
|
||||
|
||||
type Hash = u64;
|
||||
type Hash = TestHash;
|
||||
type Number = TestNumber;
|
||||
type Header = TestSourceHeader;
|
||||
type FinalityProof = TestFinalityProof;
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq)]
|
||||
struct TestSourceHeader(IsMandatory, TestNumber);
|
||||
struct TestSourceHeader(IsMandatory, TestNumber, TestHash);
|
||||
|
||||
impl SourceHeader<TestHash, TestNumber> for TestSourceHeader {
|
||||
fn hash(&self) -> TestHash {
|
||||
self.2
|
||||
}
|
||||
|
||||
impl SourceHeader<TestNumber> for TestSourceHeader {
|
||||
fn number(&self) -> TestNumber {
|
||||
self.1
|
||||
}
|
||||
@@ -90,7 +102,7 @@ struct ClientsData {
|
||||
source_headers: HashMap<TestNumber, (TestSourceHeader, Option<TestFinalityProof>)>,
|
||||
source_proofs: Vec<TestFinalityProof>,
|
||||
|
||||
target_best_block_number: TestNumber,
|
||||
target_best_block_id: HeaderId<TestHash, TestNumber>,
|
||||
target_headers: Vec<(TestSourceHeader, TestFinalityProof)>,
|
||||
}
|
||||
|
||||
@@ -152,10 +164,12 @@ impl RelayClient for TestTargetClient {
|
||||
|
||||
#[async_trait]
|
||||
impl TargetClient<TestFinalitySyncPipeline> for TestTargetClient {
|
||||
async fn best_finalized_source_block_number(&self) -> Result<TestNumber, TestError> {
|
||||
async fn best_finalized_source_block_id(
|
||||
&self,
|
||||
) -> Result<HeaderId<TestHash, TestNumber>, TestError> {
|
||||
let mut data = self.data.lock();
|
||||
(self.on_method_call)(&mut *data);
|
||||
Ok(data.target_best_block_number)
|
||||
Ok(data.target_best_block_id)
|
||||
}
|
||||
|
||||
async fn submit_finality_proof(
|
||||
@@ -165,7 +179,7 @@ impl TargetClient<TestFinalitySyncPipeline> for TestTargetClient {
|
||||
) -> Result<(), TestError> {
|
||||
let mut data = self.data.lock();
|
||||
(self.on_method_call)(&mut *data);
|
||||
data.target_best_block_number = header.number();
|
||||
data.target_best_block_id = HeaderId(header.number(), header.hash());
|
||||
data.target_headers.push((header, proof));
|
||||
Ok(())
|
||||
}
|
||||
@@ -187,7 +201,7 @@ fn prepare_test_clients(
|
||||
source_headers,
|
||||
source_proofs: vec![TestFinalityProof(12), TestFinalityProof(14)],
|
||||
|
||||
target_best_block_number: 5,
|
||||
target_best_block_id: HeaderId(5, 5),
|
||||
target_headers: vec![],
|
||||
}));
|
||||
(
|
||||
@@ -199,6 +213,15 @@ fn prepare_test_clients(
|
||||
)
|
||||
}
|
||||
|
||||
fn test_sync_params() -> FinalitySyncParams {
|
||||
FinalitySyncParams {
|
||||
tick: Duration::from_secs(0),
|
||||
recent_finality_proofs_limit: 1024,
|
||||
stall_timeout: Duration::from_secs(1),
|
||||
only_mandatory_headers: false,
|
||||
}
|
||||
}
|
||||
|
||||
fn run_sync_loop(
|
||||
state_function: impl Fn(&mut ClientsData) -> bool + Send + Sync + 'static,
|
||||
) -> ClientsData {
|
||||
@@ -207,21 +230,17 @@ fn run_sync_loop(
|
||||
exit_sender,
|
||||
state_function,
|
||||
vec![
|
||||
(6, (TestSourceHeader(false, 6), None)),
|
||||
(7, (TestSourceHeader(false, 7), Some(TestFinalityProof(7)))),
|
||||
(8, (TestSourceHeader(true, 8), Some(TestFinalityProof(8)))),
|
||||
(9, (TestSourceHeader(false, 9), Some(TestFinalityProof(9)))),
|
||||
(10, (TestSourceHeader(false, 10), None)),
|
||||
(5, (TestSourceHeader(false, 5, 5), None)),
|
||||
(6, (TestSourceHeader(false, 6, 6), None)),
|
||||
(7, (TestSourceHeader(false, 7, 7), Some(TestFinalityProof(7)))),
|
||||
(8, (TestSourceHeader(true, 8, 8), Some(TestFinalityProof(8)))),
|
||||
(9, (TestSourceHeader(false, 9, 9), Some(TestFinalityProof(9)))),
|
||||
(10, (TestSourceHeader(false, 10, 10), None)),
|
||||
]
|
||||
.into_iter()
|
||||
.collect(),
|
||||
);
|
||||
let sync_params = FinalitySyncParams {
|
||||
tick: Duration::from_secs(0),
|
||||
recent_finality_proofs_limit: 1024,
|
||||
stall_timeout: Duration::from_secs(1),
|
||||
only_mandatory_headers: false,
|
||||
};
|
||||
let sync_params = test_sync_params();
|
||||
|
||||
let clients_data = source_client.data.clone();
|
||||
let _ = async_std::task::block_on(run(
|
||||
@@ -246,38 +265,38 @@ fn finality_sync_loop_works() {
|
||||
//
|
||||
// once this ^^^ is done, we generate more blocks && read proof for blocks 12 and 14 from
|
||||
// the stream
|
||||
if data.target_best_block_number == 9 {
|
||||
if data.target_best_block_id.0 == 9 {
|
||||
data.source_best_block_number = 14;
|
||||
data.source_headers.insert(11, (TestSourceHeader(false, 11), None));
|
||||
data.source_headers.insert(11, (TestSourceHeader(false, 11, 11), None));
|
||||
data.source_headers
|
||||
.insert(12, (TestSourceHeader(false, 12), Some(TestFinalityProof(12))));
|
||||
data.source_headers.insert(13, (TestSourceHeader(false, 13), None));
|
||||
.insert(12, (TestSourceHeader(false, 12, 12), Some(TestFinalityProof(12))));
|
||||
data.source_headers.insert(13, (TestSourceHeader(false, 13, 13), None));
|
||||
data.source_headers
|
||||
.insert(14, (TestSourceHeader(false, 14), Some(TestFinalityProof(14))));
|
||||
.insert(14, (TestSourceHeader(false, 14, 14), Some(TestFinalityProof(14))));
|
||||
}
|
||||
// once this ^^^ is done, we generate more blocks && read persistent proof for block 16
|
||||
if data.target_best_block_number == 14 {
|
||||
if data.target_best_block_id.0 == 14 {
|
||||
data.source_best_block_number = 17;
|
||||
data.source_headers.insert(15, (TestSourceHeader(false, 15), None));
|
||||
data.source_headers.insert(15, (TestSourceHeader(false, 15, 15), None));
|
||||
data.source_headers
|
||||
.insert(16, (TestSourceHeader(false, 16), Some(TestFinalityProof(16))));
|
||||
data.source_headers.insert(17, (TestSourceHeader(false, 17), None));
|
||||
.insert(16, (TestSourceHeader(false, 16, 16), Some(TestFinalityProof(16))));
|
||||
data.source_headers.insert(17, (TestSourceHeader(false, 17, 17), None));
|
||||
}
|
||||
|
||||
data.target_best_block_number == 16
|
||||
data.target_best_block_id.0 == 16
|
||||
});
|
||||
|
||||
assert_eq!(
|
||||
client_data.target_headers,
|
||||
vec![
|
||||
// before adding 11..14: finality proof for mandatory header#8
|
||||
(TestSourceHeader(true, 8), TestFinalityProof(8)),
|
||||
(TestSourceHeader(true, 8, 8), TestFinalityProof(8)),
|
||||
// before adding 11..14: persistent finality proof for non-mandatory header#9
|
||||
(TestSourceHeader(false, 9), TestFinalityProof(9)),
|
||||
(TestSourceHeader(false, 9, 9), TestFinalityProof(9)),
|
||||
// after adding 11..14: ephemeral finality proof for non-mandatory header#14
|
||||
(TestSourceHeader(false, 14), TestFinalityProof(14)),
|
||||
(TestSourceHeader(false, 14, 14), TestFinalityProof(14)),
|
||||
// after adding 15..17: persistent finality proof for non-mandatory header#16
|
||||
(TestSourceHeader(false, 16), TestFinalityProof(16)),
|
||||
(TestSourceHeader(false, 16, 16), TestFinalityProof(16)),
|
||||
],
|
||||
);
|
||||
}
|
||||
@@ -291,11 +310,11 @@ fn run_only_mandatory_headers_mode_test(
|
||||
exit_sender,
|
||||
|_| false,
|
||||
vec![
|
||||
(6, (TestSourceHeader(false, 6), Some(TestFinalityProof(6)))),
|
||||
(7, (TestSourceHeader(false, 7), Some(TestFinalityProof(7)))),
|
||||
(8, (TestSourceHeader(has_mandatory_headers, 8), Some(TestFinalityProof(8)))),
|
||||
(9, (TestSourceHeader(false, 9), Some(TestFinalityProof(9)))),
|
||||
(10, (TestSourceHeader(false, 10), Some(TestFinalityProof(10)))),
|
||||
(6, (TestSourceHeader(false, 6, 6), Some(TestFinalityProof(6)))),
|
||||
(7, (TestSourceHeader(false, 7, 7), Some(TestFinalityProof(7)))),
|
||||
(8, (TestSourceHeader(has_mandatory_headers, 8, 8), Some(TestFinalityProof(8)))),
|
||||
(9, (TestSourceHeader(false, 9, 9), Some(TestFinalityProof(9)))),
|
||||
(10, (TestSourceHeader(false, 10, 10), Some(TestFinalityProof(10)))),
|
||||
]
|
||||
.into_iter()
|
||||
.collect(),
|
||||
@@ -322,7 +341,7 @@ fn select_header_to_submit_skips_non_mandatory_headers_when_only_mandatory_heade
|
||||
assert_eq!(run_only_mandatory_headers_mode_test(true, false), None);
|
||||
assert_eq!(
|
||||
run_only_mandatory_headers_mode_test(false, false),
|
||||
Some((TestSourceHeader(false, 10), TestFinalityProof(10))),
|
||||
Some((TestSourceHeader(false, 10, 10), TestFinalityProof(10))),
|
||||
);
|
||||
}
|
||||
|
||||
@@ -330,11 +349,11 @@ fn select_header_to_submit_skips_non_mandatory_headers_when_only_mandatory_heade
|
||||
fn select_header_to_submit_selects_mandatory_headers_when_only_mandatory_headers_are_required() {
|
||||
assert_eq!(
|
||||
run_only_mandatory_headers_mode_test(true, true),
|
||||
Some((TestSourceHeader(true, 8), TestFinalityProof(8))),
|
||||
Some((TestSourceHeader(true, 8, 8), TestFinalityProof(8))),
|
||||
);
|
||||
assert_eq!(
|
||||
run_only_mandatory_headers_mode_test(false, true),
|
||||
Some((TestSourceHeader(true, 8), TestFinalityProof(8))),
|
||||
Some((TestSourceHeader(true, 8, 8), TestFinalityProof(8))),
|
||||
);
|
||||
}
|
||||
|
||||
@@ -345,63 +364,74 @@ fn select_better_recent_finality_proof_works() {
|
||||
select_better_recent_finality_proof::<TestFinalitySyncPipeline>(
|
||||
&[(5, TestFinalityProof(5))],
|
||||
&mut vec![],
|
||||
Some((TestSourceHeader(false, 2), TestFinalityProof(2))),
|
||||
Some((TestSourceHeader(false, 2, 2), TestFinalityProof(2))),
|
||||
),
|
||||
Some((TestSourceHeader(false, 2), TestFinalityProof(2))),
|
||||
Some((TestSourceHeader(false, 2, 2), TestFinalityProof(2))),
|
||||
);
|
||||
|
||||
// if there are no recent finality proofs, nothing is changed
|
||||
assert_eq!(
|
||||
select_better_recent_finality_proof::<TestFinalitySyncPipeline>(
|
||||
&[],
|
||||
&mut vec![TestSourceHeader(false, 5)],
|
||||
Some((TestSourceHeader(false, 2), TestFinalityProof(2))),
|
||||
&mut vec![TestSourceHeader(false, 5, 5)],
|
||||
Some((TestSourceHeader(false, 2, 2), TestFinalityProof(2))),
|
||||
),
|
||||
Some((TestSourceHeader(false, 2), TestFinalityProof(2))),
|
||||
Some((TestSourceHeader(false, 2, 2), TestFinalityProof(2))),
|
||||
);
|
||||
|
||||
// if there's no intersection between recent finality proofs and unjustified headers, nothing is
|
||||
// changed
|
||||
let mut unjustified_headers = vec![TestSourceHeader(false, 9), TestSourceHeader(false, 10)];
|
||||
let mut unjustified_headers =
|
||||
vec![TestSourceHeader(false, 9, 9), TestSourceHeader(false, 10, 10)];
|
||||
assert_eq!(
|
||||
select_better_recent_finality_proof::<TestFinalitySyncPipeline>(
|
||||
&[(1, TestFinalityProof(1)), (4, TestFinalityProof(4))],
|
||||
&mut unjustified_headers,
|
||||
Some((TestSourceHeader(false, 2), TestFinalityProof(2))),
|
||||
Some((TestSourceHeader(false, 2, 2), TestFinalityProof(2))),
|
||||
),
|
||||
Some((TestSourceHeader(false, 2), TestFinalityProof(2))),
|
||||
Some((TestSourceHeader(false, 2, 2), TestFinalityProof(2))),
|
||||
);
|
||||
|
||||
// if there's intersection between recent finality proofs and unjustified headers, but there are
|
||||
// no proofs in this intersection, nothing is changed
|
||||
let mut unjustified_headers =
|
||||
vec![TestSourceHeader(false, 8), TestSourceHeader(false, 9), TestSourceHeader(false, 10)];
|
||||
let mut unjustified_headers = vec![
|
||||
TestSourceHeader(false, 8, 8),
|
||||
TestSourceHeader(false, 9, 9),
|
||||
TestSourceHeader(false, 10, 10),
|
||||
];
|
||||
assert_eq!(
|
||||
select_better_recent_finality_proof::<TestFinalitySyncPipeline>(
|
||||
&[(7, TestFinalityProof(7)), (11, TestFinalityProof(11))],
|
||||
&mut unjustified_headers,
|
||||
Some((TestSourceHeader(false, 2), TestFinalityProof(2))),
|
||||
Some((TestSourceHeader(false, 2, 2), TestFinalityProof(2))),
|
||||
),
|
||||
Some((TestSourceHeader(false, 2), TestFinalityProof(2))),
|
||||
Some((TestSourceHeader(false, 2, 2), TestFinalityProof(2))),
|
||||
);
|
||||
assert_eq!(
|
||||
unjustified_headers,
|
||||
vec![TestSourceHeader(false, 8), TestSourceHeader(false, 9), TestSourceHeader(false, 10)]
|
||||
vec![
|
||||
TestSourceHeader(false, 8, 8),
|
||||
TestSourceHeader(false, 9, 9),
|
||||
TestSourceHeader(false, 10, 10)
|
||||
]
|
||||
);
|
||||
|
||||
// if there's intersection between recent finality proofs and unjustified headers and there's
|
||||
// a proof in this intersection:
|
||||
// - this better (last from intersection) proof is selected;
|
||||
// - 'obsolete' unjustified headers are pruned.
|
||||
let mut unjustified_headers =
|
||||
vec![TestSourceHeader(false, 8), TestSourceHeader(false, 9), TestSourceHeader(false, 10)];
|
||||
let mut unjustified_headers = vec![
|
||||
TestSourceHeader(false, 8, 8),
|
||||
TestSourceHeader(false, 9, 9),
|
||||
TestSourceHeader(false, 10, 10),
|
||||
];
|
||||
assert_eq!(
|
||||
select_better_recent_finality_proof::<TestFinalitySyncPipeline>(
|
||||
&[(7, TestFinalityProof(7)), (9, TestFinalityProof(9))],
|
||||
&mut unjustified_headers,
|
||||
Some((TestSourceHeader(false, 2), TestFinalityProof(2))),
|
||||
Some((TestSourceHeader(false, 2, 2), TestFinalityProof(2))),
|
||||
),
|
||||
Some((TestSourceHeader(false, 9), TestFinalityProof(9))),
|
||||
Some((TestSourceHeader(false, 9, 9), TestFinalityProof(9))),
|
||||
);
|
||||
}
|
||||
|
||||
@@ -475,3 +505,45 @@ fn prune_recent_finality_proofs_works() {
|
||||
prune_recent_finality_proofs::<TestFinalitySyncPipeline>(20, &mut recent_finality_proofs, 2);
|
||||
assert_eq!(&original_recent_finality_proofs[5..], recent_finality_proofs);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn different_forks_at_source_and_at_target_are_detected() {
|
||||
let (exit_sender, _exit_receiver) = futures::channel::mpsc::unbounded();
|
||||
let (source_client, target_client) = prepare_test_clients(
|
||||
exit_sender,
|
||||
|_| false,
|
||||
vec![
|
||||
(5, (TestSourceHeader(false, 5, 42), None)),
|
||||
(6, (TestSourceHeader(false, 6, 6), None)),
|
||||
(7, (TestSourceHeader(false, 7, 7), None)),
|
||||
(8, (TestSourceHeader(false, 8, 8), None)),
|
||||
(9, (TestSourceHeader(false, 9, 9), None)),
|
||||
(10, (TestSourceHeader(false, 10, 10), None)),
|
||||
]
|
||||
.into_iter()
|
||||
.collect(),
|
||||
);
|
||||
|
||||
let mut progress = (Instant::now(), None);
|
||||
let mut finality_proofs_stream = RestartableFinalityProofsStream {
|
||||
needs_restart: false,
|
||||
stream: Box::pin(futures::stream::iter(vec![]).boxed()),
|
||||
};
|
||||
let mut recent_finality_proofs = Vec::new();
|
||||
let metrics_sync = SyncLoopMetrics::new(None, "source", "target").unwrap();
|
||||
async_std::task::block_on(run_loop_iteration::<TestFinalitySyncPipeline, _, _>(
|
||||
&source_client,
|
||||
&target_client,
|
||||
FinalityLoopState {
|
||||
progress: &mut progress,
|
||||
finality_proofs_stream: &mut finality_proofs_stream,
|
||||
recent_finality_proofs: &mut recent_finality_proofs,
|
||||
last_transaction: None,
|
||||
},
|
||||
&test_sync_params(),
|
||||
&Some(metrics_sync.clone()),
|
||||
))
|
||||
.unwrap();
|
||||
|
||||
assert!(!metrics_sync.is_using_same_fork());
|
||||
}
|
||||
|
||||
@@ -19,8 +19,9 @@
|
||||
//! are still submitted to the target node, but are treated as auxiliary data as we are not trying
|
||||
//! to submit all source headers to the target node.
|
||||
|
||||
pub use crate::finality_loop::{
|
||||
metrics_prefix, run, FinalitySyncParams, SourceClient, TargetClient,
|
||||
pub use crate::{
|
||||
finality_loop::{metrics_prefix, run, FinalitySyncParams, SourceClient, TargetClient},
|
||||
sync_loop_metrics::SyncLoopMetrics,
|
||||
};
|
||||
|
||||
use bp_header_chain::FinalityProof;
|
||||
@@ -42,13 +43,15 @@ pub trait FinalitySyncPipeline: 'static + Clone + Debug + Send + Sync {
|
||||
/// Headers we're syncing are identified by this number.
|
||||
type Number: relay_utils::BlockNumberBase;
|
||||
/// Type of header that we're syncing.
|
||||
type Header: SourceHeader<Self::Number>;
|
||||
type Header: SourceHeader<Self::Hash, Self::Number>;
|
||||
/// Finality proof type.
|
||||
type FinalityProof: FinalityProof<Self::Number>;
|
||||
}
|
||||
|
||||
/// Header that we're receiving from source node.
|
||||
pub trait SourceHeader<Number>: Clone + Debug + PartialEq + Send + Sync {
|
||||
pub trait SourceHeader<Hash, Number>: Clone + Debug + PartialEq + Send + Sync {
|
||||
/// Returns hash of header.
|
||||
fn hash(&self) -> Hash;
|
||||
/// Returns number of header.
|
||||
fn number(&self) -> Number;
|
||||
/// Returns true if this header needs to be submitted to target node.
|
||||
|
||||
@@ -16,49 +16,71 @@
|
||||
|
||||
//! Metrics for headers synchronization relay loop.
|
||||
|
||||
use relay_utils::metrics::{
|
||||
metric_name, register, GaugeVec, Metric, Opts, PrometheusError, Registry, U64,
|
||||
};
|
||||
use relay_utils::metrics::{metric_name, register, IntGauge, Metric, PrometheusError, Registry};
|
||||
|
||||
/// Headers sync metrics.
|
||||
#[derive(Clone)]
|
||||
pub struct SyncLoopMetrics {
|
||||
/// Best syncing headers at "source" and "target" nodes.
|
||||
best_block_numbers: GaugeVec<U64>,
|
||||
/// Best syncing header at the source.
|
||||
best_source_block_number: IntGauge,
|
||||
/// Best syncing header at the target.
|
||||
best_target_block_number: IntGauge,
|
||||
/// Flag that has `0` value when best source headers at the source node and at-target-chain
|
||||
/// are matching and `1` otherwise.
|
||||
using_different_forks: IntGauge,
|
||||
}
|
||||
|
||||
impl SyncLoopMetrics {
|
||||
/// Create and register headers loop metrics.
|
||||
pub fn new(prefix: Option<&str>) -> Result<Self, PrometheusError> {
|
||||
pub fn new(
|
||||
prefix: Option<&str>,
|
||||
at_source_chain_label: &str,
|
||||
at_target_chain_label: &str,
|
||||
) -> Result<Self, PrometheusError> {
|
||||
Ok(SyncLoopMetrics {
|
||||
best_block_numbers: GaugeVec::new(
|
||||
Opts::new(
|
||||
metric_name(prefix, "best_block_numbers"),
|
||||
"Best block numbers on source and target nodes",
|
||||
),
|
||||
&["node"],
|
||||
best_source_block_number: IntGauge::new(
|
||||
metric_name(prefix, &format!("best_{}_block_number", at_source_chain_label)),
|
||||
format!("Best block number at the {}", at_source_chain_label),
|
||||
)?,
|
||||
best_target_block_number: IntGauge::new(
|
||||
metric_name(prefix, &format!("best_{}_block_number", at_target_chain_label)),
|
||||
format!("Best block number at the {}", at_target_chain_label),
|
||||
)?,
|
||||
using_different_forks: IntGauge::new(
|
||||
metric_name(prefix, &format!("is_{}_and_{}_using_different_forks", at_source_chain_label, at_target_chain_label)),
|
||||
"Whether the best finalized source block at target node is different (value 1) from the \
|
||||
corresponding block at the source node",
|
||||
)?,
|
||||
})
|
||||
}
|
||||
|
||||
/// Returns current value of the using-same-fork flag.
|
||||
#[cfg(test)]
|
||||
pub(crate) fn is_using_same_fork(&self) -> bool {
|
||||
self.using_different_forks.get() == 0
|
||||
}
|
||||
|
||||
/// Update best block number at source.
|
||||
pub fn update_best_block_at_source<Number: Into<u64>>(&self, source_best_number: Number) {
|
||||
self.best_block_numbers
|
||||
.with_label_values(&["source"])
|
||||
.set(source_best_number.into());
|
||||
self.best_source_block_number.set(source_best_number.into());
|
||||
}
|
||||
|
||||
/// Update best block number at target.
|
||||
pub fn update_best_block_at_target<Number: Into<u64>>(&self, target_best_number: Number) {
|
||||
self.best_block_numbers
|
||||
.with_label_values(&["target"])
|
||||
.set(target_best_number.into());
|
||||
self.best_target_block_number.set(target_best_number.into());
|
||||
}
|
||||
|
||||
/// Update using-same-fork flag.
|
||||
pub fn update_using_same_fork(&self, using_same_fork: bool) {
|
||||
self.using_different_forks.set(if using_same_fork { 0 } else { 1 })
|
||||
}
|
||||
}
|
||||
|
||||
impl Metric for SyncLoopMetrics {
|
||||
fn register(&self, registry: &Registry) -> Result<(), PrometheusError> {
|
||||
register(self.best_block_numbers.clone(), registry)?;
|
||||
register(self.best_source_block_number.clone(), registry)?;
|
||||
register(self.best_target_block_number.clone(), registry)?;
|
||||
register(self.using_different_forks.clone(), registry)?;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user