pezkuwichain/pezkuwi-subxt
1
0
Fork 0
mirror of https://github.com/pezkuwichain/pezkuwi-subxt.git synced 2026-05-31 01:41:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

Squashed commit of slava-async-bridge:

Browse Source 
commit 657deb4cf4b90f24b9c5bfd62764b197776c262c
Author: Hernando Castano <castano.ha@gmail.com>
Date:   Wed Jan 29 20:14:20 2020 -0500

    Move Slava's bridge code into relays folder

commit 4868c42c7da959dde7252766996b3ed4e408e439
Author: Hernando Castano <castano.ha@gmail.com>
Date:   Wed Jan 29 20:01:06 2020 -0500

    Move files into `modules/ethereum`

commit d1093f3e4238acb1a1a020011452cb928d3f8d7a
Merge: 29dc6f9 bfd30ef
Author: Hernando Castano <castano.ha@gmail.com>
Date:   Wed Jan 29 19:59:27 2020 -0500

    Merge branch 'master' of slava-async-bridge

commit 29dc6f97b1b7d1db99086d35a5336f43d2f0f8af
Author: Hernando Castano <castano.ha@gmail.com>
Date:   Wed Jan 29 19:51:31 2020 -0500

    Squashed commit of the following:

    commit 265365920836bb1d286c9b48b1902a2de278fdd9
    Author: Hernando Castano <castano.ha@gmail.com>
    Date:   Wed Jan 29 19:51:15 2020 -0500

        Move hc-jp-bridge repo to different folder

    commit 8271991e95320baba70bd1cb9c4234d0ffd5b638
    Merge: 57d0811 304cbc5
    Author: Hernando Castano <castano.ha@gmail.com>
    Date:   Wed Jan 29 19:36:41 2020 -0500

        Merge branch 'hc-jp-bridge-module' of hc-jp-bridge-module

    commit 304cbc5f02d003ffa5404c1c01e461e5b8539888
    Author: Hernando Castano <HCastano@users.noreply.github.com>
    Date:   Wed Jan 29 00:38:27 2020 -0500

        Update bridge pallet to work with the (almost) lastest master (#4672)

        * Update decl_error usage

        * WIP: Update error handling to use DispatchResult

        * Get module compiling with new error handling

        * Make tests compile again

        Main change was updating the usage of InMemoryBackend

        * Move `sp-state-machine` into dev-dependencies

        * Bump dependencies to v2.0.0

        * Remove some stray comments

        * Appy code review suggestion

    commit 510cd6d96372688517496efa61773ea2839f8474
    Author: Hernando Castano <HCastano@users.noreply.github.com>
    Date:   Tue Dec 17 12:52:51 2019 -0500

        Move Bridge Pallet into FRAME (#4373)

        * Move `bridge` crate into `frame` folder

        * Make `bridge` pallet compile after `the-big-reorg`

    commit ab54e838ef75e6a3f68fd0944bf22598c10c552f
    Author: Hernando Castano <castano.ha@gmail.com>
    Date:   Mon Nov 11 21:56:40 2019 +0100

        Use new StorageProof type from #3834

    commit 8fc8911fd1b4acc2274c6863fb3dba91b30c90af
    Author: Hernando Castano <HCastano@users.noreply.github.com>
    Date:   Tue Nov 5 00:50:34 2019 +0100

        Verify Ancestry between Headers (#3963)

        * Create module for checking ancestry proofs

        * Use Vec of Headers instead of a HashMap

        * Move the ancestry verification into the lib.rs file

        * Change the proof format to exclude `child` and `ancestor` headers

        * Add a testing function for building header chains

        * Rename AncestorNotFound error to InvalidAncestryProof

        * Use ancestor hash instead of header when verifying ancestry

        * Clean up some stuff missed in the merge

    commit dbe85738b68358b790cf927b34a804b965a88f96
    or: Hernando Castano <HCastano@users.noreply.github.com>
    Date:   Fri Nov 1 15:41:58 2019 +0100

        Check given Grandpa validator set against set found in storage (#3915)

        * Make StorageProofChecker happy

        * Update some tests

        * Check given validator set against set found in storage

        * Use Finality Grandpa's Authority Id and Weight

        * Add better error handling

        * Use error type from decl_error! macro

    commit 31b09216603d3e9c21144ce8c0b6bf59307a4f97
    or: Hernando Castano <HCastano@users.noreply.github.com>
    Date:   Wed Oct 23 14:55:37 2019 +0200

        Make tests work after the changes introduced in #3793 (#3874)

        * Make tests work after the changes introduced in #3793

        * Remove unneccessary import

    commit bce6d804aa86504599ff912387295c58f846cbf3
    Author: Jim Posen <jim.posen@gmail.com>
    Date:   Thu Oct 10 12:18:58 2019 +0200

        Logic for checking Substrate proofs from within runtime module. (#3783)

    commit a7013e94b6c772c1d45a7cacbb445f73f6554fca
    Author: Hernando Castano <castano.ha@gmail.com>
    Date:   Fri Oct 4 15:21:00 2019 +0300

        Allow tracking of multiple bridges

    commit 3cf648242d631e32bd553a67df54bf5a48912839
    Author: Hernando Castano <castano.ha@gmail.com>
    Date:   Tue Oct 1 14:55:04 2019 +0200

        Add BridgeId => Bridge mapping

    commit 001c74c45072213e01857d0a2454379b447c5a76
    Author: Hernando Castano <castano.ha@gmail.com>
    Date:   Tue Oct 1 11:10:19 2019 +0200

        Get the mock runtime for tests set up

    commit 38443a1e8b424ed2f148eb95121d009f730e3b5a
    Author: Hernando Castano <castano.ha@gmail.com>
    Date:   Fri Sep 27 14:52:53 2019 +0200

        Clean up some warnings

    commit bdc3b01401e89c7111f8bf71f84c50750d25089f
    Author: Hernando Castano <castano.ha@gmail.com>
    Date:   Thu Sep 26 16:41:01 2019 +0200

        Add more skeleton code

    commit 26995efbf4bac2842eb2822322f7ad3c3e88feb8
    Author: Hernando Castano <castano.ha@gmail.com>
    Date:   Wed Sep 25 15:16:57 2019 +0200

        Create `bridge` module skeleton

commit bfd30ef8363b1483ef1107ae1eb958a4e944c93b
Author: Svyatoslav Nikolsky <svyatonik@gmail.com>
Date:   Tue Dec 10 12:10:53 2019 +0300

    actually use signer from CLI to sign Substrate transactions

commit 504028eac60d9d14ba95b506cd355b0d2f405ce0
Author: Svyatoslav Nikolsky <svyatonik@gmail.com>
Date:   Tue Dec 10 12:02:22 2019 +0300

    go offline for a bit on connection error

commit 446d0c8d20187dfd1beb173958ea28f2ad97887d
Author: Svyatoslav Nikolsky <svyatonik@gmail.com>
Date:   Tue Dec 10 11:25:50 2019 +0300

    enable info logs by default

commit d039c60ec72bc91adfdad85442bc99a93b7f8e8d
Author: Svyatoslav Nikolsky <svyatonik@gmail.com>
Date:   Tue Dec 10 11:12:51 2019 +0300

    support basic CLI arguments

commit 65c6d48e23576f36e8541878b920a03730226392
Author: Svyatoslav Nikolsky <svyatonik@gmail.com>
Date:   Mon Dec 9 15:37:48 2019 +0300

    fix restart

commit 96e94c1c4b22d732078f8c401b872c5f8246c3fe
Author: Svyatoslav Nikolsky <svyatonik@gmail.com>
Date:   Mon Dec 9 14:57:53 2019 +0300

    license

commit 68f4191e6cdd211ac8975e0b79f8a6f46a3ca953
Author: Svyatoslav Nikolsky <svyatonik@gmail.com>
Date:   Mon Dec 9 14:56:05 2019 +0300

    restart sync when Substrate reorgs && we are unlucky

commit 29887c446167d580d73cc03a0b71c31890cafb51
Author: Svyatoslav Nikolsky <svyatonik@gmail.com>
Date:   Mon Dec 9 13:49:31 2019 +0300

    only read genesis hash once

commit 832492b8393fe2063adf9c58c2b9e060dc3e4efb
Author: Svyatoslav Nikolsky <svyatonik@gmail.com>
Date:   Mon Dec 9 13:23:26 2019 +0300

    changed TODO

commit 9dbc130e5fa036ae63d973819daf30f4ed6ffb5b
Author: Svyatoslav Nikolsky <svyatonik@gmail.com>
Date:   Mon Dec 9 13:16:56 2019 +0300

    removed obsolete exit future

commit d03408cd8284eb0c61e7e96429b4f6199353e030
Author: Svyatoslav Nikolsky <svyatonik@gmail.com>
Date:   Mon Dec 9 13:16:17 2019 +0300

    removed obsolete TODOs + moved a couple of TODOs to runtime module

commit ed8bec44b79f9a2ce829e59f10181368b2f42139
Author: Svyatoslav Nikolsky <svyatonik@gmail.com>
Date:   Mon Dec 9 12:37:05 2019 +0300

    explained TODO fix

commit aa9c4c66ec2904eeb6072d654718b0ac0b7d8803
Author: Svyatoslav Nikolsky <svyatonik@gmail.com>
Date:   Mon Dec 9 12:28:09 2019 +0300

    fix tx outcome serialization

commit 126f8f5484dac8c4af588ae86dc8855919d6c822
Author: Svyatoslav Nikolsky <svyatonik@gmail.com>
Date:   Mon Dec 9 12:05:05 2019 +0300

    prune old ethereum headers when Substrate best header is too far in the future

commit c7bd301e631a44fe3263e188d0956081aa84f31e
Author: Svyatoslav Nikolsky <svyatonik@gmail.com>
Date:   Fri Dec 6 12:51:50 2019 +0300

    fix trace

commit 549bb7acdb30cfdafe6c8600f0410212539ea63d
Author: Svyatoslav Nikolsky <svyatonik@gmail.com>
Date:   Fri Dec 6 12:51:26 2019 +0300

    tx hashes are already a part of Block response

commit 7864017909f87ea36955d605a924c3c88bc88df3
Author: Svyatoslav Nikolsky <svyatonik@gmail.com>
Date:   Thu Dec 5 12:29:37 2019 +0300

    submit bunch of headers at once + some fixes

commit 96485f85d38c144f0771f02ba692216a60356665
Author: Svyatoslav Nikolsky <svyatonik@gmail.com>
Date:   Wed Dec 4 17:22:13 2019 +0300

    print status messages

commit ae0ec4c087136db653339537daab7f96a8c21b65
Author: Svyatoslav Nikolsky <svyatonik@gmail.com>
Date:   Wed Dec 4 17:06:00 2019 +0300

    continue actual Substrate client implementation

commit 8146293740d70b88904568ff8e5acdfbadf06fd3
Author: Svyatoslav Nikolsky <svyatonik@gmail.com>
Date:   Wed Dec 4 13:49:30 2019 +0300

    fix IncompleteHeader condition

commit 767c6201157dabcccf7f62e643681ca298224fb1
Author: Svyatoslav Nikolsky <svyatonik@gmail.com>
Date:   Wed Dec 4 10:55:06 2019 +0300

    actual Substrate client implementation

commit 221fd4ccd2b1eea12c9dacf800d80e15ec115c1b
Author: Svyatoslav Nikolsky <svyatonik@gmail.com>
Date:   Wed Nov 20 17:28:13 2019 +0300

    initial commit
This commit is contained in:
Hernando Castano
2020-01-29 20:14:48 -05:00
committed by Bastian Köcher
parent d530bf2199
commit 479c5bd99a
13 changed files with 9349 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bridges/relays/ethereum/src/ethereum_sync_loop.rs
+486
View File
@@ -0,0 +1,486 @@
// Copyright 2019 Parity Technologies (UK) Ltd.
// This file is part of Parity-Bridge.
// Parity-Bridge is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// Parity-Bridge is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
// You should have received a copy of the GNU General Public License
// along with Parity-Bridge. If not, see <http://www.gnu.org/licenses/>.
use futures::{future::FutureExt, stream::StreamExt};
use crate::ethereum_client;
use crate::ethereum_types::HeaderStatus as EthereumHeaderStatus;
use crate::substrate_client;
// TODO: when SharedClient will be available, switch to Substrate headers subscription
// (because we do not need old Substrate headers)
/// Interval (in ms) at which we check new Ethereum headers when we are synced/almost synced.
const ETHEREUM_TICK_INTERVAL_MS: u64 = 10_000;
/// Interval (in ms) at which we check new Substrate blocks.
const SUBSTRATE_TICK_INTERVAL_MS: u64 = 5_000;
/// When we submit Ethereum headers to Substrate runtime, but see no updates of best
/// Ethereum block known to Substrate runtime during STALL_SYNC_TIMEOUT_MS milliseconds,
/// we consider that our headers are rejected because there has been reorg in Substrate.
/// This reorg could invalidate our knowledge about sync process (i.e. we have asked if
/// HeaderA is known to Substrate, but then reorg happened and the answer is different
/// now) => we need to reset sync.
/// The other option is to receive **EVERY** best Substrate header and check if it is
/// direct child of previous best header. But: (1) subscription doesn't guarantee that
/// the subscriber will receive every best header (2) reorg won't always lead to sync
/// stall and restart is a heavy operation (we forget all in-memory headers).
const STALL_SYNC_TIMEOUT_MS: u64 = 30_000;
/// Delay (in milliseconds) after connection-related error happened before we'll try
/// reconnection again.
const CONNECTION_ERROR_DELAY_MS: u64 = 10_000;
/// Error type that can signal connection errors.
pub trait MaybeConnectionError {
/// Returns true if error (maybe) represents connection error.
fn is_connection_error(&self) -> bool;
}
/// Ethereum synchronization parameters.
pub struct EthereumSyncParams {
/// Ethereum RPC host.
pub eth_host: String,
/// Ethereum RPC port.
pub eth_port: u16,
/// Substrate RPC host.
pub sub_host: String,
/// Substrate RPC port.
pub sub_port: u16,
/// Substrate transactions signer.
pub sub_signer: sp_core::sr25519::Pair,
/// Maximal number of ethereum headers to pre-download.
pub max_future_headers_to_download: usize,
/// Maximal number of active (we believe) submit header transactions.
pub max_headers_in_submitted_status: usize,
/// Maximal number of headers in single submit request.
pub max_headers_in_single_submit: usize,
/// Maximal total headers size in single submit request.
pub max_headers_size_in_single_submit: usize,
/// We only may store and accept (from Ethereum node) headers that have
/// number >= than best_substrate_header.number - prune_depth.
pub prune_depth: u64,
}
impl std::fmt::Debug for EthereumSyncParams {
fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
f.debug_struct("EthereumSyncParams")
.field("eth_host", &self.eth_host)
.field("eth_port", &self.eth_port)
.field("sub_host", &self.sub_port)
.field("sub_port", &self.sub_port)
.field("max_future_headers_to_download", &self.max_future_headers_to_download)
.field("max_headers_in_submitted_status", &self.max_headers_in_submitted_status)
.field("max_headers_in_single_submit", &self.max_headers_in_single_submit)
.field("max_headers_size_in_single_submit", &self.max_headers_size_in_single_submit)
.field("prune_depth", &self.prune_depth)
.finish()
}
}
impl Default for EthereumSyncParams {
fn default() -> Self {
EthereumSyncParams {
eth_host: "localhost".into(),
eth_port: 8545,
sub_host: "localhost".into(),
sub_port: 9933,
sub_signer: sp_keyring::AccountKeyring::Alice.pair(),
max_future_headers_to_download: 128,
max_headers_in_submitted_status: 128,
max_headers_in_single_submit: 32,
max_headers_size_in_single_submit: 131_072,
prune_depth: 4096,
}
}
}
/// Run Ethereum headers synchronization.
pub fn run(params: EthereumSyncParams) {
let mut local_pool = futures::executor::LocalPool::new();
let mut progress_context = (std::time::Instant::now(), None, None);
local_pool.run_until(async move {
let eth_uri = format!("http://{}:{}", params.eth_host, params.eth_port);
let sub_uri = format!("http://{}:{}", params.sub_host, params.sub_port);
let sub_signer = params.sub_signer.clone();
let mut eth_sync = crate::ethereum_sync::HeadersSync::new(params);
let mut stall_countdown = None;
let mut eth_maybe_client = None;
let mut eth_best_block_number_required = false;
let eth_best_block_number_future = ethereum_client::best_block_number(
ethereum_client::client(&eth_uri)
).fuse();
let eth_new_header_future = futures::future::Fuse::terminated();
let eth_orphan_header_future = futures::future::Fuse::terminated();
let eth_receipts_future = futures::future::Fuse::terminated();
let eth_go_offline_future = futures::future::Fuse::terminated();
let eth_tick_stream = interval(ETHEREUM_TICK_INTERVAL_MS).fuse();
let mut sub_maybe_client = None;
let mut sub_best_block_required = false;
let sub_best_block_future = substrate_client::best_ethereum_block(
substrate_client::client(&sub_uri, sub_signer),
).fuse();
let sub_receipts_check_future = futures::future::Fuse::terminated();
let sub_existence_status_future = futures::future::Fuse::terminated();
let sub_submit_header_future = futures::future::Fuse::terminated();
let sub_go_offline_future = futures::future::Fuse::terminated();
let sub_tick_stream = interval(SUBSTRATE_TICK_INTERVAL_MS).fuse();
futures::pin_mut!(
eth_best_block_number_future,
eth_new_header_future,
eth_orphan_header_future,
eth_receipts_future,
eth_go_offline_future,
eth_tick_stream,
sub_best_block_future,
sub_receipts_check_future,
sub_existence_status_future,
sub_submit_header_future,
sub_go_offline_future,
sub_tick_stream
);
loop {
futures::select! {
(eth_client, eth_best_block_number) = eth_best_block_number_future => {
eth_best_block_number_required = false;
process_future_result(
&mut eth_maybe_client,
eth_client,
eth_best_block_number,
|eth_best_block_number| eth_sync.ethereum_best_header_number_response(eth_best_block_number),
&mut eth_go_offline_future,
|eth_client| delay(CONNECTION_ERROR_DELAY_MS, eth_client),
"Error retrieving best header number from Ethereum number",
);
},
(eth_client, eth_new_header) = eth_new_header_future => {
process_future_result(
&mut eth_maybe_client,
eth_client,
eth_new_header,
|eth_new_header| eth_sync.headers_mut().header_response(eth_new_header),
&mut eth_go_offline_future,
|eth_client| delay(CONNECTION_ERROR_DELAY_MS, eth_client),
"Error retrieving header from Ethereum node",
);
},
(eth_client, eth_orphan_header) = eth_orphan_header_future => {
process_future_result(
&mut eth_maybe_client,
eth_client,
eth_orphan_header,
|eth_orphan_header| eth_sync.headers_mut().header_response(eth_orphan_header),
&mut eth_go_offline_future,
|eth_client| delay(CONNECTION_ERROR_DELAY_MS, eth_client),
"Error retrieving orphan header from Ethereum node",
);
},
(eth_client, eth_receipts) = eth_receipts_future => {
process_future_result(
&mut eth_maybe_client,
eth_client,
eth_receipts,
|(header, receipts)| eth_sync.headers_mut().receipts_response(&header, receipts),
&mut eth_go_offline_future,
|eth_client| delay(CONNECTION_ERROR_DELAY_MS, eth_client),
"Error retrieving transactions receipts from Ethereum node",
);
},
eth_client = eth_go_offline_future => {
eth_maybe_client = Some(eth_client);
},
_ = eth_tick_stream.next() => {
if eth_sync.is_almost_synced() {
eth_best_block_number_required = true;
}
},
(sub_client, sub_best_block) = sub_best_block_future => {
sub_best_block_required = false;
process_future_result(
&mut sub_maybe_client,
sub_client,
sub_best_block,
|sub_best_block| {
let head_updated = eth_sync.substrate_best_header_response(sub_best_block);
match head_updated {
// IF head is updated AND there are still our transactions:
// => restart stall countdown timer
true if eth_sync.headers().headers_in_status(EthereumHeaderStatus::Submitted) != 0 =>
stall_countdown = Some(std::time::Instant::now()),
// IF head is updated AND there are no our transactions:
// => stop stall countdown timer
true => stall_countdown = None,
// IF head is not updated AND stall countdown is not yet completed
// => do nothing
false if stall_countdown
.map(|stall_countdown| std::time::Instant::now() - stall_countdown <
std::time::Duration::from_millis(STALL_SYNC_TIMEOUT_MS))
.unwrap_or(true)
=> (),
// IF head is not updated AND stall countdown has completed
// => restart sync
false => {
log::info!(
target: "bridge",
"Possible Substrate fork detected. Restarting Ethereum headers synchronization.",
);
stall_countdown = None;
eth_sync.restart();
},
}
},
&mut sub_go_offline_future,
|sub_client| delay(CONNECTION_ERROR_DELAY_MS, sub_client),
"Error retrieving best known header from Substrate node",
);
},
(sub_client, sub_existence_status) = sub_existence_status_future => {
process_future_result(
&mut sub_maybe_client,
sub_client,
sub_existence_status,
|(sub_header, sub_existence_status)| eth_sync
.headers_mut()
.maybe_orphan_response(&sub_header, sub_existence_status),
&mut sub_go_offline_future,
|sub_client| delay(CONNECTION_ERROR_DELAY_MS, sub_client),
"Error retrieving existence status from Substrate node",
);
},
(sub_client, sub_submit_header_result) = sub_submit_header_future => {
process_future_result(
&mut sub_maybe_client,
sub_client,
sub_submit_header_result,
|(_, submitted_headers)| eth_sync.headers_mut().headers_submitted(submitted_headers),
&mut sub_go_offline_future,
|sub_client| delay(CONNECTION_ERROR_DELAY_MS, sub_client),
"Error submitting headers to Substrate node",
);
},
(sub_client, sub_receipts_check_result) = sub_receipts_check_future => {
// we can minimize number of receipts_check calls by checking header
// logs bloom here, but it may give us false positives (when authorities
// source is contract, we never need any logs)
process_future_result(
&mut sub_maybe_client,
sub_client,
sub_receipts_check_result,
|(header, receipts_check_result)| eth_sync
.headers_mut()
.maybe_receipts_response(&header, receipts_check_result),
&mut sub_go_offline_future,
|sub_client| delay(CONNECTION_ERROR_DELAY_MS, sub_client),
"Error retrieving receipts requirement from Substrate node",
);
},
sub_client = sub_go_offline_future => {
sub_maybe_client = Some(sub_client);
},
_ = sub_tick_stream.next() => {
sub_best_block_required = true;
},
}
// print progress
progress_context = print_progress(progress_context, &eth_sync);
// if client is available: wait, or call Substrate RPC methods
if let Some(sub_client) = sub_maybe_client.take() {
// the priority is to:
// 1) get best block - it stops us from downloading/submitting new blocks + we call it rarely;
// 2) check transactions receipts - it stops us from downloading/submitting new blocks;
// 3) check existence - it stops us from submitting new blocks;
// 4) submit header
if sub_best_block_required {
log::debug!(target: "bridge", "Asking Substrate about best block");
sub_best_block_future.set(substrate_client::best_ethereum_block(sub_client).fuse());
} else if let Some(header) = eth_sync.headers().header(EthereumHeaderStatus::MaybeReceipts) {
log::debug!(
target: "bridge",
"Checking if header submission requires receipts: {:?}",
header.id(),
);
let header = header.clone();
sub_receipts_check_future.set(
substrate_client::ethereum_receipts_required(sub_client, header).fuse()
);
} else if let Some(header) = eth_sync.headers().header(EthereumHeaderStatus::MaybeOrphan) {
// for MaybeOrphan we actually ask for parent' header existence
let parent_id = header.parent_id();
log::debug!(
target: "bridge",
"Asking Substrate node for existence of: {:?}",
parent_id,
);
sub_existence_status_future.set(
substrate_client::ethereum_header_known(sub_client, parent_id).fuse(),
);
} else if let Some(headers) = eth_sync.select_headers_to_submit() {
let ids = match headers.len() {
1 => format!("{:?}", headers[0].id()),
2 => format!("[{:?}, {:?}]", headers[0].id(), headers[1].id()),
len => format!("[{:?} ... {:?}]", headers[0].id(), headers[len - 1].id()),
};
log::debug!(
target: "bridge",
"Submitting {} header(s) to Substrate node: {:?}",
headers.len(),
ids,
);
let headers = headers.into_iter().cloned().collect();
sub_submit_header_future.set(
substrate_client::submit_ethereum_headers(sub_client, headers).fuse(),
);
// remember that we have submitted some headers
if stall_countdown.is_none() {
stall_countdown = Some(std::time::Instant::now());
}
} else {
sub_maybe_client = Some(sub_client);
}
}
// if client is available: wait, or call Ethereum RPC methods
if let Some(eth_client) = eth_maybe_client.take() {
// the priority is to:
// 1) get best block - it stops us from downloading new blocks + we call it rarely;
// 2) check transactions receipts - it stops us from downloading/submitting new blocks;
// 3) check existence - it stops us from submitting new blocks;
// 4) submit header
if eth_best_block_number_required {
log::debug!(target: "bridge", "Asking Ethereum node about best block number");
eth_best_block_number_future.set(ethereum_client::best_block_number(eth_client).fuse());
} else if let Some(header) = eth_sync.headers().header(EthereumHeaderStatus::Receipts) {
let id = header.id();
log::debug!(
target: "bridge",
"Retrieving receipts for header: {:?}",
id,
);
eth_receipts_future.set(
ethereum_client::transactions_receipts(
eth_client,
id,
header.header().transactions.clone(),
).fuse()
);
} else if let Some(header) = eth_sync.headers().header(EthereumHeaderStatus::Orphan) {
// for Orphan we actually ask for parent' header
let parent_id = header.parent_id();
log::debug!(
target: "bridge",
"Going to download orphan header from Ethereum node: {:?}",
parent_id,
);
eth_orphan_header_future.set(
ethereum_client::header_by_hash(eth_client, parent_id.1).fuse(),
);
} else if let Some(id) = eth_sync.select_new_header_to_download() {
log::debug!(
target: "bridge",
"Going to download new header from Ethereum node: {:?}",
id,
);
eth_new_header_future.set(
ethereum_client::header_by_number(eth_client, id).fuse(),
);
} else {
eth_maybe_client = Some(eth_client);
}
}
}
});
}
fn print_progress(
progress_context: (std::time::Instant, Option<u64>, Option<u64>),
eth_sync: &crate::ethereum_sync::HeadersSync,
) -> (std::time::Instant, Option<u64>, Option<u64>) {
let (prev_time, prev_best_header, prev_target_header) = progress_context;
let now_time = std::time::Instant::now();
let (now_best_header, now_target_header) = eth_sync.status();
let need_update = now_time - prev_time > std::time::Duration::from_secs(10)
|| match (prev_best_header, now_best_header) {
(Some(prev_best_header), Some(now_best_header)) => now_best_header.0.saturating_sub(prev_best_header) > 10,
_ => false,
};
if !need_update {
return (prev_time, prev_best_header, prev_target_header);
}
log::info!(
target: "bridge",
"Synced {:?} of {:?} headers",
now_best_header.map(|id| id.0),
now_target_header,
);
(now_time, now_best_header.clone().map(|id| id.0), *now_target_header)
}
async fn delay<T>(timeout_ms: u64, retval: T) -> T {
async_std::task::sleep(std::time::Duration::from_millis(timeout_ms)).await;
retval
}
fn interval(timeout_ms: u64) -> impl futures::Stream<Item = ()> {
futures::stream::unfold((), move |_| async move { delay(timeout_ms, ()).await; Some(((), ())) })
}
fn process_future_result<TClient, TResult, TError, TGoOfflineFuture>(
maybe_client: &mut Option<TClient>,
client: TClient,
result: Result<TResult, TError>,
on_success: impl FnOnce(TResult),
go_offline_future: &mut std::pin::Pin<&mut futures::future::Fuse<TGoOfflineFuture>>,
go_offline: impl FnOnce(TClient) -> TGoOfflineFuture,
error_pattern: &'static str,
) where
TError: std::fmt::Debug + MaybeConnectionError,
TGoOfflineFuture: FutureExt,
{
match result {
Ok(result) => {
*maybe_client = Some(client);
on_success(result);
},
Err(error) => {
if error.is_connection_error() {
go_offline_future.set(go_offline(client).fuse());
} else {
*maybe_client = Some(client);
}
log::error!(target: "bridge", "{}: {:?}", error_pattern, error);
},
}
}