Solidity contract that accepts unverified substrate headers (#65)

* solidity contract

* continue

* upd

* cargo update

* fixes

* ehtereum_headers -> headers

* extracted some common stuff

* ethereum_sync.rs -> sync.rs

* make sync generic

* continue extracting

* continue

* add eth-contract argument

* continue

* some fixes

* contract v2

* continue

* more fixes

* more fixes

* deal with duplicated params

* removed multiple call_rpc variants

* bail_on_error!()

* fn submit_ethereum_transaction

* more fixes

* cargo fmt --all

* fix

* bail_on_arg_error!()

* fix

* fix

* remove async_extra stuff

* substrate-bridge.json -> substrate-bridge-abi.json

* get rid of substrate transactions hashes

* get rid of ethereum transactions hashes

* extracted contract bytecode to separate file

* cargo fmt --all

* avoid duplicate import in contracts

* removed Default::default()

* swapped configurations for sub2eth && eth2sub

* fix compilation

* do not double gas limit when submitting Substrate headers

* cargo fmt --all

* solidity contract removed

* consts

* extracted solc compilation details to separate file

* removed (obsolete in future Vec<u8> justification)

* fixed cli option description

* fix typos

* fix grumble

* extracted constants

* log decoded header

* cargo fmt --all

* comment

bridges/modules/ethereum-contract/builtin/Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "ethereum-contract-builtin"
+description = "Small crate that helps Solidity contract to verify finality proof."
+version = "0.1.0"
+authors = ["Parity Technologies <admin@parity.io>"]
+edition = "2018"
+
+[dependencies]
+
+# General dependencies
+
+codec = { package = "parity-scale-codec", version = "1.0.0" }
+finality-grandpa = { version = "0.11.2", features = ["derive-codec"] }
+sp-blockchain = "2.0.0-alpha.5"
+sp-finality-grandpa = "2.0.0-alpha.5"
+sp-runtime = "2.0.0-alpha.5"
+
+# Runtime/chain specific dependencies
+
+bridge-node-runtime = { path = "../../../bin/node/runtime" }

bridges/modules/ethereum-contract/builtin/src/lib.rs

@@ -0,0 +1,79 @@
+// Copyright 2019-2020 Parity Technologies (UK) Ltd.
+// This file is part of Parity Bridges Common.
+
+// Parity Bridges Common 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 Bridges Common 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 Bridges Common.  If not, see <http://www.gnu.org/licenses/>.
+
+use bridge_node_runtime::{BlockNumber, Hash, Header as RuntimeHeader};
+use codec::Decode;
+use sp_blockchain::Error as ClientError;
+
+/// Builtin errors.
+#[derive(Debug)]
+pub enum Error {
+	/// Failed to decode Substrate header.
+	HeaderDecode(codec::Error),
+	/// Failed to decode best voters set.
+	BestVotersDecode(codec::Error),
+	/// Failed to decode finality proof.
+	FinalityProofDecode(codec::Error),
+	/// Failed to verify justification.
+	JustificationVerify(ClientError),
+}
+
+/// Substrate header.
+#[derive(Debug)]
+pub struct Header {
+	/// Header hash.
+	pub hash: Hash,
+	/// Parent header hash.
+	pub parent_hash: Hash,
+	/// Header number.
+	pub number: BlockNumber,
+	/// GRANDPA validators change signal.
+	pub signal: Option<ValidatorsSetSignal>,
+}
+
+/// GRANDPA validators set change signal.
+#[derive(Debug)]
+pub struct ValidatorsSetSignal {
+	/// Signal delay.
+	pub delay: BlockNumber,
+	/// New validators set.
+	pub validators: Vec<u8>,
+}
+
+/// Parse Substrate header.
+pub fn parse_substrate_header(raw_header: &[u8]) -> Result<Header, Error> {
+	RuntimeHeader::decode(&mut &raw_header[..])
+		.map(|header| Header {
+			hash: header.hash(),
+			parent_hash: header.parent_hash,
+			number: header.number,
+			signal: None, // TODO: parse me
+		})
+		.map_err(Error::HeaderDecode)
+}
+
+/// Verify GRANDPA finality proof.
+pub fn verify_substrate_finality_proof(
+	_best_set_id: u64,
+	_raw_best_voters: &[u8],
+	_raw_best_header: &[u8],
+	_raw_headers: &[&[u8]],
+	_raw_finality_proof: &[u8],
+) -> Result<(usize, usize), Error> {
+	Err(Error::JustificationVerify(ClientError::Msg(
+		"Not yet implemented".into(),
+	))) // TODO: implement me
+}

bridges/relays/ethereum/Cargo.toml

@@ -11,11 +11,17 @@ async-stream = "0.2.0"
 clap = { version = "2.33.0", features = ["yaml"] }
 codec = { package = "parity-scale-codec", version = "1.0.0" }
 env_logger = "0.7.0"
+ethabi = "12.0"
+ethabi-contract = "11.0"
+ethabi-derive = "11.0"
+ethereum-tx-sign = { git = "https://github.com/svyatonik/ethereum-tx-sign.git", branch = "up-ethereum-types" }
 futures = "0.3.1"
+hex = "0.4"
 jsonrpsee = { git = "https://github.com/paritytech/jsonrpsee.git", default-features = false, features = ["http"] }
 linked-hash-map = "0.5.2"
 log = "0.4.8"
 num-traits = "0.2"
+parity-crypto = { version = "0.6", features = ["publickey"] }
 parking_lot = "0.10.2"
 rustc-hex = "2.0.1"
 serde = { version = "1.0.106", features = ["derive"] }

bridges/relays/ethereum/res/substrate-bridge-abi.json

@@ -0,0 +1,119 @@
+[
+	{
+		"inputs": [
+			{
+				"internalType": "bytes",
+				"name": "rawInitialHeader",
+				"type": "bytes"
+			},
+			{
+				"internalType": "uint64",
+				"name": "initialValidatorsSetId",
+				"type": "uint64"
+			},
+			{
+				"internalType": "bytes",
+				"name": "initialValidatorsSet",
+				"type": "bytes"
+			}
+		],
+		"stateMutability": "nonpayable",
+		"type": "constructor"
+	},
+	{
+		"stateMutability": "nonpayable",
+		"type": "fallback"
+	},
+	{
+		"inputs": [],
+		"name": "bestKnownHeader",
+		"outputs": [
+			{
+				"internalType": "uint256",
+				"name": "",
+				"type": "uint256"
+			},
+			{
+				"internalType": "bytes32",
+				"name": "",
+				"type": "bytes32"
+			}
+		],
+		"stateMutability": "view",
+		"type": "function"
+	},
+	{
+		"inputs": [
+			{
+				"internalType": "uint256",
+				"name": "finalityTargetNumber",
+				"type": "uint256"
+			},
+			{
+				"internalType": "bytes32",
+				"name": "finalityTargetHash",
+				"type": "bytes32"
+			},
+			{
+				"internalType": "bytes",
+				"name": "rawFinalityProof",
+				"type": "bytes"
+			}
+		],
+		"name": "importFinalityProof",
+		"outputs": [],
+		"stateMutability": "nonpayable",
+		"type": "function"
+	},
+	{
+		"inputs": [
+			{
+				"internalType": "bytes",
+				"name": "rawHeader",
+				"type": "bytes"
+			}
+		],
+		"name": "importHeader",
+		"outputs": [],
+		"stateMutability": "nonpayable",
+		"type": "function"
+	},
+	{
+		"inputs": [
+			{
+				"internalType": "bytes32",
+				"name": "headerHash",
+				"type": "bytes32"
+			}
+		],
+		"name": "isFinalityProofRequired",
+		"outputs": [
+			{
+				"internalType": "bool",
+				"name": "",
+				"type": "bool"
+			}
+		],
+		"stateMutability": "view",
+		"type": "function"
+	},
+	{
+		"inputs": [
+			{
+				"internalType": "bytes32",
+				"name": "headerHash",
+				"type": "bytes32"
+			}
+		],
+		"name": "isKnownHeader",
+		"outputs": [
+			{
+				"internalType": "bool",
+				"name": "",
+				"type": "bool"
+			}
+		],
+		"stateMutability": "view",
+		"type": "function"
+	}
+]

bridges/relays/ethereum/res/substrate-bridge-metadata.txt

@@ -0,0 +1,5 @@
+Last Change Date: 2020-04-28
+Solc version: 0.6.6+commit.6c089d02
+Source hash (keccak256): 0xdc46aff04e37129265223e507d17f1407a70cb1ecea3230e1eaa77a17586724d
+Source gist: https://gist.github.com/svyatonik/876b388f9507a8de242cb2db9547c4f0
+Compiler flags used (command to produce the file): `docker run -i ethereum/solc:0.6.6 --optimize --bin - < substrate-bridge.sol`

bridges/relays/ethereum/src/cli.yml

@@ -43,3 +43,82 @@ subcommands:
                 long: sub-signer-password
                 value_name: SUB_SIGNER_PASSWORD
                 help: The password for the SURI of secret key to use when transactions are submitted to the Substrate node.
+    - sub-to-eth:
+        about: Synchronize headers from Substrate node to Ethereum node.
+        args:
+            - eth-host:
+                long: eth-host
+                value_name: ETH_HOST
+                help: Connect to Ethereum node at given host.
+                takes_value: true
+            - eth-port:
+                long: eth-port
+                value_name: ETH_PORT
+                help: Connect to Ethereum node at given port.
+                takes_value: true
+            - eth-contract:
+                long: eth-contract
+                value_name: ETH_CONTRACT
+                help: Address of deployed bridge contract.
+                takes_value: true
+            - eth-signer:
+                long: eth-signer
+                value_name: ETH_SIGNER
+                help: Hex-encoded secret to use when transactions are submitted to the Ethereum node.
+            - sub-host:
+                long: sub-host
+                value_name: SUB_HOST
+                help: Connect to Substrate node at given host.
+                takes_value: true
+            - sub-port:
+                long: sub-port
+                value_name: SUB_PORT
+                help: Connect to Substrate node at given port.
+                takes_value: true
+    - eth-deploy-contract:
+        about: Deploy Bridge contract on Ethereum node.
+        args:
+            - eth-host:
+                long: eth-host
+                value_name: ETH_HOST
+                help: Connect to Ethereum node at given host.
+                takes_value: true
+            - eth-port:
+                long: eth-port
+                value_name: ETH_PORT
+                help: Connect to Ethereum node at given port.
+                takes_value: true
+            - eth-signer:
+                long: eth-signer
+                value_name: ETH_SIGNER
+                help: Hex-encoded secret to use when transactions are submitted to the Ethereum node.
+            - eth-contract-code:
+                long: eth-contract-code
+                value_name: ETH_CONTRACT_CODE
+                help: Bytecode of bridge contract.
+                takes_value: true
+            - sub-host:
+                long: sub-host
+                value_name: SUB_HOST
+                help: Connect to Substrate node at given host.
+                takes_value: true
+            - sub-port:
+                long: sub-port
+                value_name: SUB_PORT
+                help: Connect to Substrate node at given port.
+                takes_value: true
+            - sub-authorities-set-id:
+                long: sub-authorities-set-id
+                value_name: SUB_AUTHORITIES_SET_ID
+                help: ID of initial GRANDPA authorities set.
+                takes_value: true
+            - sub-authorities-set:
+                long: sub-authorities-set
+                value_name: SUB_AUTHORITIES_SET
+                help: Encoded initial GRANDPA authorities set.
+                takes_value: true
+            - sub-initial-header:
+                long: sub-initial-header
+                value_name: SUB_INITIAL_HEADER
+                help: Encoded initial Substrate header.
+                takes_value: true

bridges/relays/ethereum/src/ethereum_client.rs

@@ -14,14 +14,22 @@
 // You should have received a copy of the GNU General Public License
 // along with Parity Bridges Common.  If not, see <http://www.gnu.org/licenses/>.
 
-use crate::ethereum_types::{EthereumHeaderId, Header, Receipt, H256, U64};
+use crate::ethereum_types::{Address, Bytes, EthereumHeaderId, Header, Receipt, TransactionHash, H256, U256, U64};
-use crate::sync_types::MaybeConnectionError;
+use crate::substrate_types::{Hash as SubstrateHash, QueuedSubstrateHeader, SubstrateHeaderId};
+use crate::sync_types::{HeaderId, MaybeConnectionError};
+use crate::{bail_on_arg_error, bail_on_error};
+use codec::{Decode, Encode};
+use ethabi::FunctionOutputDecoder;
 use jsonrpsee::common::Params;
 use jsonrpsee::raw::{RawClient, RawClientError};
 use jsonrpsee::transport::http::{HttpTransportClient, RequestError};
-use serde::de::DeserializeOwned;
+use parity_crypto::publickey::KeyPair;
+use serde::{de::DeserializeOwned, Serialize};
 use serde_json::{from_value, to_value};
 
+// to encode/decode contract calls
+ethabi_contract::use_contract!(bridge_contract, "res/substrate-bridge-abi.json");
+
 /// Proof of hash serialization success.
 const HASH_SERIALIZATION_PROOF: &'static str = "hash serialization never fails; qed";
 /// Proof of integer serialization success.
@@ -29,24 +37,83 @@ const INT_SERIALIZATION_PROOF: &'static str = "integer serialization never fails
 /// Proof of bool serialization success.
 const BOOL_SERIALIZATION_PROOF: &'static str = "bool serialization never fails; qed";
 
+/// Ethereum connection params.
+#[derive(Debug)]
+pub struct EthereumConnectionParams {
+	/// Ethereum RPC host.
+	pub host: String,
+	/// Ethereum RPC port.
+	pub port: u16,
+}
+
+impl Default for EthereumConnectionParams {
+	fn default() -> Self {
+		EthereumConnectionParams {
+			host: "localhost".into(),
+			port: 8545,
+		}
+	}
+}
+
+/// Ethereum signing params.
+#[derive(Clone, Debug)]
+pub struct EthereumSigningParams {
+	/// Ethereum chain id.
+	pub chain_id: u64,
+	/// Ethereum transactions signer.
+	pub signer: KeyPair,
+	/// Gas price we agree to pay.
+	pub gas_price: U256,
+}
+
+impl Default for EthereumSigningParams {
+	fn default() -> Self {
+		EthereumSigningParams {
+			chain_id: 0x11, // Parity dev chain
+			// account that has a lot of ether when we run instant seal engine
+			// address: 0x00a329c0648769a73afac7f9381e08fb43dbea72
+			// secret: 0x4d5db4107d237df6a3d58ee5f70ae63d73d7658d4026f2eefd2f204c81682cb7
+			signer: KeyPair::from_secret_slice(
+				&hex::decode("4d5db4107d237df6a3d58ee5f70ae63d73d7658d4026f2eefd2f204c81682cb7")
+					.expect("secret is hardcoded, thus valid; qed"),
+			)
+			.expect("secret is hardcoded, thus valid; qed"),
+			gas_price: 8_000_000_000u64.into(), // 8 Gwei
+		}
+	}
+}
+
 /// Ethereum client type.
 pub type Client = RawClient<HttpTransportClient>;
 
+/// Ethereum contract call request.
+#[derive(Debug, Default, PartialEq, Serialize)]
+pub struct CallRequest {
+	/// Contract address.
+	pub to: Option<Address>,
+	/// Call data.
+	pub data: Option<Bytes>,
+}
+
 /// All possible errors that can occur during interacting with Ethereum node.
 #[derive(Debug)]
 pub enum Error {
 	/// Request start failed.
 	StartRequestFailed(RequestError),
+	/// Error serializing request.
+	RequestSerialization(serde_json::Error),
 	/// Request not found (should never occur?).
 	RequestNotFound,
 	/// Failed to receive response.
 	ResponseRetrievalFailed(RawClientError<RequestError>),
 	/// Failed to parse response.
-	ResponseParseFailed(serde_json::Error),
+	ResponseParseFailed(String),
 	/// We have received header with missing number and hash fields.
 	IncompleteHeader,
 	/// We have received receipt with missing gas_used field.
 	IncompleteReceipt,
+	/// Invalid Substrate block number received from Ethereum node.
+	InvalidSubstrateBlockNumber,
 }
 
 impl MaybeConnectionError for Error {
@@ -59,8 +126,9 @@ impl MaybeConnectionError for Error {
 }
 
 /// Returns client that is able to call RPCs on Ethereum node.
-pub fn client(uri: &str) -> Client {
+pub fn client(params: EthereumConnectionParams) -> Client {
-	let transport = HttpTransportClient::new(uri);
+	let uri = format!("http://{}:{}", params.host, params.port);
+	let transport = HttpTransportClient::new(&uri);
 	RawClient::new(transport)
 }
 
@@ -118,15 +186,11 @@ pub async fn header_by_hash(client: Client, hash: H256) -> (Client, Result<Heade
 pub async fn transactions_receipts(
 	mut client: Client,
 	id: EthereumHeaderId,
-	transacactions: Vec<H256>,
+	transactions: Vec<H256>,
 ) -> (Client, Result<(EthereumHeaderId, Vec<Receipt>), Error>) {
-	let mut transactions_receipts = Vec::with_capacity(transacactions.len());
+	let mut transactions_receipts = Vec::with_capacity(transactions.len());
-	for transacaction in transacactions {
+	for transaction in transactions {
-		let (next_client, transaction_receipt) = transaction_receipt(client, transacaction).await;
+		let (next_client, transaction_receipt) = bail_on_error!(transaction_receipt(client, transaction).await);
-		let transaction_receipt = match transaction_receipt {
-			Ok(transaction_receipt) => transaction_receipt,
-			Err(error) => return (next_client, Err(error)),
-		};
 		transactions_receipts.push(transaction_receipt);
 		client = next_client;
 	}
@@ -150,6 +214,172 @@ async fn transaction_receipt(client: Client, hash: H256) -> (Client, Result<Rece
 	)
 }
 
+/// Returns best Substrate block that PoA chain knows of.
+pub async fn best_substrate_block(
+	client: Client,
+	contract_address: Address,
+) -> (Client, Result<SubstrateHeaderId, Error>) {
+	let (encoded_call, call_decoder) = bridge_contract::functions::best_known_header::call();
+	let call_request = bail_on_arg_error!(
+		to_value(CallRequest {
+			to: Some(contract_address),
+			data: Some(encoded_call.into()),
+		})
+		.map_err(|e| Error::RequestSerialization(e)),
+		client
+	);
+	let (client, call_result) =
+		bail_on_error!(call_rpc::<Bytes>(client, "eth_call", Params::Array(vec![call_request]),).await);
+	let (number, raw_hash) = match call_decoder.decode(&call_result.0) {
+		Ok((raw_number, raw_hash)) => (raw_number, raw_hash),
+		Err(error) => return (client, Err(Error::ResponseParseFailed(format!("{}", error)))),
+	};
+	let hash = match SubstrateHash::decode(&mut &raw_hash[..]) {
+		Ok(hash) => hash,
+		Err(error) => return (client, Err(Error::ResponseParseFailed(format!("{}", error)))),
+	};
+
+	if number != number.low_u32().into() {
+		return (client, Err(Error::InvalidSubstrateBlockNumber));
+	}
+
+	(client, Ok(HeaderId(number.low_u32(), hash)))
+}
+
+/// Returns true if Substrate header is known to Ethereum node.
+pub async fn substrate_header_known(
+	client: Client,
+	contract_address: Address,
+	id: SubstrateHeaderId,
+) -> (Client, Result<(SubstrateHeaderId, bool), Error>) {
+	let (encoded_call, call_decoder) = bridge_contract::functions::is_known_header::call(id.1);
+	let call_request = bail_on_arg_error!(
+		to_value(CallRequest {
+			to: Some(contract_address),
+			data: Some(encoded_call.into()),
+		})
+		.map_err(|e| Error::RequestSerialization(e)),
+		client
+	);
+	let (client, call_result) =
+		bail_on_error!(call_rpc::<Bytes>(client, "eth_call", Params::Array(vec![call_request]),).await);
+	match call_decoder.decode(&call_result.0) {
+		Ok(is_known_block) => (client, Ok((id, is_known_block))),
+		Err(error) => (client, Err(Error::ResponseParseFailed(format!("{}", error)))),
+	}
+}
+
+/// Submits Substrate headers to Ethereum contract.
+pub async fn submit_substrate_headers(
+	client: Client,
+	params: EthereumSigningParams,
+	contract_address: Address,
+	headers: Vec<QueuedSubstrateHeader>,
+) -> (Client, Result<Vec<SubstrateHeaderId>, Error>) {
+	let (mut client, mut nonce) =
+		bail_on_error!(account_nonce(client, params.signer.address().as_fixed_bytes().into()).await);
+
+	let ids = headers.iter().map(|header| header.id()).collect();
+	for header in headers {
+		client = bail_on_error!(
+			submit_ethereum_transaction(
+				client,
+				&params,
+				Some(contract_address),
+				Some(nonce),
+				false,
+				bridge_contract::functions::import_header::encode_input(header.extract().0.encode(),),
+			)
+			.await
+		)
+		.0;
+
+		nonce += 1.into();
+	}
+
+	(client, Ok(ids))
+}
+
+/// Deploy bridge contract.
+pub async fn deploy_bridge_contract(
+	client: Client,
+	params: &EthereumSigningParams,
+	contract_code: Vec<u8>,
+	initial_header: Vec<u8>,
+	initial_set_id: u64,
+	initial_authorities: Vec<u8>,
+) -> (Client, Result<(), Error>) {
+	submit_ethereum_transaction(
+		client,
+		params,
+		None,
+		None,
+		false,
+		bridge_contract::constructor(contract_code, initial_header, initial_set_id, initial_authorities),
+	)
+	.await
+}
+
+/// Submit ethereum transaction.
+async fn submit_ethereum_transaction(
+	client: Client,
+	params: &EthereumSigningParams,
+	contract_address: Option<Address>,
+	nonce: Option<U256>,
+	double_gas: bool,
+	encoded_call: Vec<u8>,
+) -> (Client, Result<(), Error>) {
+	let (client, nonce) = match nonce {
+		Some(nonce) => (client, nonce),
+		None => bail_on_error!(account_nonce(client, params.signer.address().as_fixed_bytes().into()).await),
+	};
+	let (client, gas) = bail_on_error!(
+		estimate_gas(
+			client,
+			CallRequest {
+				to: contract_address,
+				data: Some(encoded_call.clone().into()),
+			}
+		)
+		.await
+	);
+	let raw_transaction = ethereum_tx_sign::RawTransaction {
+		nonce,
+		to: contract_address,
+		value: U256::zero(),
+		gas: if double_gas { gas.saturating_mul(2.into()) } else { gas },
+		gas_price: params.gas_price,
+		data: encoded_call,
+	}
+	.sign(&params.signer.secret().as_fixed_bytes().into(), &params.chain_id);
+	let transaction = bail_on_arg_error!(
+		to_value(Bytes(raw_transaction)).map_err(|e| Error::RequestSerialization(e)),
+		client
+	);
+	let (client, _) = bail_on_error!(
+		call_rpc::<TransactionHash>(client, "eth_submitTransaction", Params::Array(vec![transaction])).await
+	);
+	(client, Ok(()))
+}
+
+/// Get account nonce.
+async fn account_nonce(client: Client, caller_address: Address) -> (Client, Result<U256, Error>) {
+	let caller_address = bail_on_arg_error!(
+		to_value(caller_address).map_err(|e| Error::RequestSerialization(e)),
+		client
+	);
+	call_rpc(client, "eth_getTransactionCount", Params::Array(vec![caller_address])).await
+}
+
+/// Estimate gas usage for call.
+async fn estimate_gas(client: Client, call_request: CallRequest) -> (Client, Result<U256, Error>) {
+	let call_request = bail_on_arg_error!(
+		to_value(call_request).map_err(|e| Error::RequestSerialization(e)),
+		client
+	);
+	call_rpc(client, "eth_estimateGas", Params::Array(vec![call_request])).await
+}
+
 /// Calls RPC on Ethereum node.
 async fn call_rpc<T: DeserializeOwned>(
 	mut client: Client,
@@ -172,7 +402,7 @@ async fn call_rpc<T: DeserializeOwned>(
 			.ok_or(Error::RequestNotFound)?
 			.await
 			.map_err(Error::ResponseRetrievalFailed)?;
-		from_value(response).map_err(Error::ResponseParseFailed)
+		from_value(response).map_err(|e| Error::ResponseParseFailed(format!("{}", e)))
 	}
 
 	let result = do_call_rpc(&mut client, method, params).await;

bridges/relays/ethereum/src/ethereum_deploy_contract.rs

@@ -0,0 +1,136 @@
+// Copyright 2019-2020 Parity Technologies (UK) Ltd.
+// This file is part of Parity Bridges Common.
+
+// Parity Bridges Common 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 Bridges Common 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 Bridges Common.  If not, see <http://www.gnu.org/licenses/>.
+
+use crate::ethereum_client::{self, EthereumConnectionParams, EthereumSigningParams};
+use crate::substrate_client::{self, SubstrateConnectionParams};
+use crate::substrate_types::{Hash as SubstrateHash, Header as SubstrateHeader};
+use codec::{Decode, Encode};
+use num_traits::Zero;
+
+/// Ethereum synchronization parameters.
+#[derive(Debug)]
+pub struct EthereumDeployContractParams {
+	/// Ethereum connection params.
+	pub eth: EthereumConnectionParams,
+	/// Ethereum signing params.
+	pub eth_sign: EthereumSigningParams,
+	/// Ethereum contract bytecode.
+	pub eth_contract_code: Vec<u8>,
+	/// Substrate connection params.
+	pub sub: SubstrateConnectionParams,
+	/// Initial authorities set id.
+	pub sub_initial_authorities_set_id: Option<u64>,
+	/// Initial authorities set.
+	pub sub_initial_authorities_set: Option<Vec<u8>>,
+	/// Initial header.
+	pub sub_initial_header: Option<Vec<u8>>,
+}
+
+impl Default for EthereumDeployContractParams {
+	fn default() -> Self {
+		EthereumDeployContractParams {
+			eth: Default::default(),
+			eth_sign: Default::default(),
+			eth_contract_code: hex::decode(include_str!("../res/substrate-bridge-bytecode.hex"))
+				.expect("code is hardcoded, thus valid; qed"),
+			sub: Default::default(),
+			sub_initial_authorities_set_id: None,
+			sub_initial_authorities_set: None,
+			sub_initial_header: None,
+		}
+	}
+}
+
+/// Deploy Bridge contract on Ethereum chain.
+pub fn run(params: EthereumDeployContractParams) {
+	let mut local_pool = futures::executor::LocalPool::new();
+
+	let result = local_pool.run_until(async move {
+		let eth_client = ethereum_client::client(params.eth);
+		let sub_client = substrate_client::client(params.sub);
+
+		let (sub_client, initial_header) = prepare_initial_header(sub_client, params.sub_initial_header).await;
+		let (initial_header_hash, initial_header) = initial_header?;
+		let initial_set_id = params.sub_initial_authorities_set_id.unwrap_or(0);
+		let (_, initial_set) = prepare_initial_authorities_set(
+			sub_client,
+			initial_header_hash,
+			params.sub_initial_authorities_set,
+		).await;
+		let initial_set = initial_set?;
+
+		log::info!(
+			target: "bridge",
+			"Deploying Ethereum contract.\r\n\tInitial header: {:?}\r\n\tInitial header encoded: {}\r\n\tInitial authorities set ID: {}\r\n\tInitial authorities set: {}",
+			initial_header,
+			hex::encode(&initial_header),
+			initial_set_id,
+			hex::encode(&initial_set),
+		);
+
+		ethereum_client::deploy_bridge_contract(
+			eth_client,
+			&params.eth_sign,
+			params.eth_contract_code,
+			initial_header,
+			initial_set_id,
+			initial_set,
+		).await.1.map_err(|error| format!("Error deploying contract: {:?}", error))
+	});
+
+	if let Err(error) = result {
+		log::error!(target: "bridge", "{}", error);
+	}
+}
+
+/// Prepare initial header.
+async fn prepare_initial_header(
+	sub_client: substrate_client::Client,
+	sub_initial_header: Option<Vec<u8>>,
+) -> (substrate_client::Client, Result<(SubstrateHash, Vec<u8>), String>) {
+	match sub_initial_header {
+		Some(raw_initial_header) => match SubstrateHeader::decode(&mut &raw_initial_header[..]) {
+			Ok(initial_header) => (sub_client, Ok((initial_header.hash(), raw_initial_header))),
+			Err(error) => (sub_client, Err(format!("Error decoding initial header: {}", error))),
+		},
+		None => {
+			let (sub_client, initial_header) = substrate_client::header_by_number(sub_client, Zero::zero()).await;
+			(
+				sub_client,
+				initial_header
+					.map(|header| (header.hash(), header.encode()))
+					.map_err(|error| format!("Error reading Substrate genesis header: {:?}", error)),
+			)
+		}
+	}
+}
+
+/// Prepare initial GRANDPA authorities set.
+async fn prepare_initial_authorities_set(
+	sub_client: substrate_client::Client,
+	sub_initial_header_hash: SubstrateHash,
+	sub_initial_authorities_set: Option<Vec<u8>>,
+) -> (substrate_client::Client, Result<Vec<u8>, String>) {
+	let (sub_client, initial_authorities_set) = match sub_initial_authorities_set {
+		Some(initial_authorities_set) => (sub_client, Ok(initial_authorities_set)),
+		None => substrate_client::grandpa_authorities_set(sub_client, sub_initial_header_hash).await,
+	};
+
+	(
+		sub_client,
+		initial_authorities_set.map_err(|error| format!("Error reading GRANDPA authorities set: {:?}", error)),
+	)
+}

bridges/relays/ethereum/src/ethereum_sync_loop.rs

@@ -14,57 +14,58 @@
 // You should have received a copy of the GNU General Public License
 // along with Parity Bridges Common.  If not, see <http://www.gnu.org/licenses/>.
 
-use crate::ethereum_client;
+use crate::ethereum_client::{self, EthereumConnectionParams};
 use crate::ethereum_types::{EthereumHeaderId, EthereumHeadersSyncPipeline, Header, QueuedEthereumHeader, Receipt};
-use crate::substrate_client;
+use crate::substrate_client::{self, SubstrateConnectionParams, SubstrateSigningParams};
 use crate::sync::{HeadersSyncParams, TargetTransactionMode};
 use crate::sync_loop::{SourceClient, TargetClient};
 use futures::future::FutureExt;
 use std::{future::Future, pin::Pin};
-pub use web3::types::H256;
+use web3::types::H256;
 
 /// 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;
+/// Max number of headers in single submit transaction.
+const MAX_HEADERS_IN_SINGLE_SUBMIT: usize = 32;
+/// Max total size of headers in single submit transaction. This only affects signed
+/// submissions, when several headers are submitted at once. 4096 is the maximal **expected**
+/// size of the Ethereum header + transactions receipts (if they're required).
+const MAX_HEADERS_SIZE_IN_SINGLE_SUBMIT: usize = MAX_HEADERS_IN_SINGLE_SUBMIT * 4096;
+/// Max Ethereum headers we want to have in all 'before-submitted' states.
+const MAX_FUTURE_HEADERS_TO_DOWNLOAD: usize = 128;
+/// Max Ethereum headers count we want to have in 'submitted' state.
+const MAX_SUBMITTED_HEADERS: usize = 128;
+/// Max depth of in-memory headers in all states. Past this depth they will be forgotten (pruned).
+const PRUNE_DEPTH: u32 = 4096;
 
 /// Ethereum synchronization parameters.
 pub struct EthereumSyncParams {
-	/// Ethereum RPC host.
+	/// Ethereum connection params.
-	pub eth_host: String,
+	pub eth: EthereumConnectionParams,
-	/// Ethereum RPC port.
+	/// Substrate connection params.
-	pub eth_port: u16,
+	pub sub: SubstrateConnectionParams,
-	/// Substrate RPC host.
+	/// Substrate signing params.
-	pub sub_host: String,
+	pub sub_sign: SubstrateSigningParams,
-	/// Substrate RPC port.
-	pub sub_port: u16,
-	/// Substrate transactions signer.
-	pub sub_signer: sp_core::sr25519::Pair,
 	/// Synchronization parameters.
 	pub sync_params: HeadersSyncParams,
 }
 
-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("sync_params", &self.sync_params)
-			.finish()
-	}
-}
-
 impl Default for EthereumSyncParams {
 	fn default() -> Self {
 		EthereumSyncParams {
-			eth_host: "localhost".into(),
+			eth: Default::default(),
-			eth_port: 8545,
+			sub: Default::default(),
-			sub_host: "localhost".into(),
+			sub_sign: Default::default(),
-			sub_port: 9933,
+			sync_params: HeadersSyncParams {
-			sub_signer: sp_keyring::AccountKeyring::Alice.pair(),
+				max_future_headers_to_download: MAX_FUTURE_HEADERS_TO_DOWNLOAD,
-			sync_params: Default::default(),
+				max_headers_in_submitted_status: MAX_SUBMITTED_HEADERS,
+				max_headers_in_single_submit: MAX_HEADERS_IN_SINGLE_SUBMIT,
+				max_headers_size_in_single_submit: MAX_HEADERS_SIZE_IN_SINGLE_SUBMIT,
+				prune_depth: PRUNE_DEPTH,
+				target_tx_mode: TargetTransactionMode::Signed,
+			},
 		}
 	}
 }
@@ -112,10 +113,10 @@ impl SourceClient<EthereumHeadersSyncPipeline> for EthereumHeadersSource {
 struct SubstrateHeadersTarget {
 	/// Substrate node client.
 	client: substrate_client::Client,
-	/// Substrate transactions signer.
-	signer: sp_core::sr25519::Pair,
 	/// Whether we want to submit signed (true), or unsigned (false) transactions.
 	sign_transactions: bool,
+	/// Substrate signing params.
+	sign_params: SubstrateSigningParams,
 }
 
 impl TargetClient<EthereumHeadersSyncPipeline> for SubstrateHeadersTarget {
@@ -126,14 +127,14 @@ impl TargetClient<EthereumHeadersSyncPipeline> for SubstrateHeadersTarget {
 	type SubmitHeadersFuture = Pin<Box<dyn Future<Output = (Self, Result<Vec<EthereumHeaderId>, Self::Error>)>>>;
 
 	fn best_header_id(self) -> Self::BestHeaderIdFuture {
-		let (signer, sign_transactions) = (self.signer, self.sign_transactions);
+		let (sign_transactions, sign_params) = (self.sign_transactions, self.sign_params);
 		substrate_client::best_ethereum_block(self.client)
 			.map(move |(client, result)| {
 				(
 					SubstrateHeadersTarget {
 						client,
-						signer,
 						sign_transactions,
+						sign_params,
 					},
 					result,
 				)
@@ -142,14 +143,14 @@ impl TargetClient<EthereumHeadersSyncPipeline> for SubstrateHeadersTarget {
 	}
 
 	fn is_known_header(self, id: EthereumHeaderId) -> Self::IsKnownHeaderFuture {
-		let (signer, sign_transactions) = (self.signer, self.sign_transactions);
+		let (sign_transactions, sign_params) = (self.sign_transactions, self.sign_params);
 		substrate_client::ethereum_header_known(self.client, id)
 			.map(move |(client, result)| {
 				(
 					SubstrateHeadersTarget {
 						client,
-						signer,
 						sign_transactions,
+						sign_params,
 					},
 					result,
 				)
@@ -161,14 +162,14 @@ impl TargetClient<EthereumHeadersSyncPipeline> for SubstrateHeadersTarget {
 		// 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)
-		let (signer, sign_transactions) = (self.signer, self.sign_transactions);
+		let (sign_transactions, sign_params) = (self.sign_transactions, self.sign_params);
 		substrate_client::ethereum_receipts_required(self.client, header.clone())
 			.map(move |(client, result)| {
 				(
 					SubstrateHeadersTarget {
 						client,
-						signer,
 						sign_transactions,
+						sign_params,
 					},
 					result,
 				)
@@ -177,16 +178,16 @@ impl TargetClient<EthereumHeadersSyncPipeline> for SubstrateHeadersTarget {
 	}
 
 	fn submit_headers(self, headers: Vec<QueuedEthereumHeader>) -> Self::SubmitHeadersFuture {
-		let (signer, sign_transactions) = (self.signer, self.sign_transactions);
+		let (sign_transactions, sign_params) = (self.sign_transactions, self.sign_params);
-		substrate_client::submit_ethereum_headers(self.client, signer.clone(), headers, sign_transactions)
+		substrate_client::submit_ethereum_headers(self.client, sign_params.clone(), headers, sign_transactions)
 			.map(move |(client, result)| {
 				(
 					SubstrateHeadersTarget {
 						client,
-						signer,
 						sign_transactions,
+						sign_params,
 					},
-					result.map(|(_, submitted_headers)| submitted_headers),
+					result,
 				)
 			})
 			.boxed()
@@ -195,12 +196,9 @@ impl TargetClient<EthereumHeadersSyncPipeline> for SubstrateHeadersTarget {
 
 /// Run Ethereum headers synchronization.
 pub fn run(params: EthereumSyncParams) {
-	let eth_uri = format!("http://{}:{}", params.eth_host, params.eth_port);
+	let eth_client = ethereum_client::client(params.eth);
-	let eth_client = ethereum_client::client(&eth_uri);
+	let sub_client = substrate_client::client(params.sub);
 
-	let sub_uri = format!("http://{}:{}", params.sub_host, params.sub_port);
-	let sub_client = substrate_client::client(&sub_uri);
-	let sub_signer = params.sub_signer;
 	let sign_sub_transactions = match params.sync_params.target_tx_mode {
 		TargetTransactionMode::Signed | TargetTransactionMode::Backup => true,
 		TargetTransactionMode::Unsigned => false,
@@ -211,8 +209,8 @@ pub fn run(params: EthereumSyncParams) {
 		ETHEREUM_TICK_INTERVAL_MS,
 		SubstrateHeadersTarget {
 			client: sub_client,
-			signer: sub_signer,
 			sign_transactions: sign_sub_transactions,
+			sign_params: params.sub_sign,
 		},
 		SUBSTRATE_TICK_INTERVAL_MS,
 		params.sync_params,

bridges/relays/ethereum/src/ethereum_types.rs

@@ -18,7 +18,7 @@ use crate::substrate_types::{into_substrate_ethereum_header, into_substrate_ethe
 use crate::sync_types::{HeaderId, HeadersSyncPipeline, QueuedHeader, SourceHeader};
 use codec::Encode;
 
-pub use web3::types::{Bytes, H256, U128, U64};
+pub use web3::types::{Address, Bytes, H256, U128, U256, U64};
 
 /// When header is just received from the Ethereum node, we check that it has
 /// both number and hash fields filled.
@@ -28,6 +28,9 @@ pub const HEADER_ID_PROOF: &'static str = "checked on retrieval; qed";
 /// gas_used field filled.
 pub const RECEIPT_GAS_USED_PROOF: &'static str = "checked on retrieval; qed";
 
+/// Ethereum transaction hash type.
+pub type TransactionHash = H256;
+
 /// Ethereum header type.
 pub type Header = web3::types::Block<H256>;
 

bridges/relays/ethereum/src/headers.rs

@@ -117,7 +117,10 @@ impl<P: HeadersSyncPipeline> QueuedHeaders<P> {
 					self.maybe_extra.keys().next_back().cloned().unwrap_or_else(Zero::zero),
 					std::cmp::max(
 						self.extra.keys().next_back().cloned().unwrap_or_else(Zero::zero),
-						self.ready.keys().next_back().cloned().unwrap_or_else(Zero::zero),
+						std::cmp::max(
+							self.ready.keys().next_back().cloned().unwrap_or_else(Zero::zero),
+							self.submitted.keys().next_back().cloned().unwrap_or_else(Zero::zero),
+						),
 					),
 				),
 			),
@@ -226,6 +229,7 @@ impl<P: HeadersSyncPipeline> QueuedHeaders<P> {
 		// all ancestors of this header are now synced => let's remove them from
 		// queues
 		let mut current = *id;
+		let mut id_processed = false;
 		loop {
 			let header = match self.status(&current) {
 				HeaderStatus::Unknown => break,
@@ -253,22 +257,26 @@ impl<P: HeadersSyncPipeline> QueuedHeaders<P> {
 				.entry(current.1)
 				.or_insert(HeaderStatus::Synced) = HeaderStatus::Synced;
 			current = header.parent_id();
+			id_processed = true;
 		}
 
 		// remember that the header is synced
-		log::debug!(
+		if !id_processed {
-			target: "bridge",
+			// to avoid duplicate log message
-			"{} header {:?} is now {:?}",
+			log::debug!(
-			P::SOURCE_NAME,
+				target: "bridge",
-			id,
+				"{} header {:?} is now {:?}",
-			HeaderStatus::Synced,
+				P::SOURCE_NAME,
-		);
+				id,
-		*self
+				HeaderStatus::Synced,
-			.known_headers
+			);
-			.entry(id.0)
+			*self
-			.or_default()
+				.known_headers
-			.entry(id.1)
+				.entry(id.0)
-			.or_insert(HeaderStatus::Synced) = HeaderStatus::Synced;
+				.or_default()
+				.entry(id.1)
+				.or_insert(HeaderStatus::Synced) = HeaderStatus::Synced;
+		}
 
 		// now let's move all descendants from maybe_orphan && orphan queues to
 		// maybe_extra queue

bridges/relays/ethereum/src/main.rs

@@ -17,17 +17,25 @@
 #![recursion_limit = "1024"]
 
 mod ethereum_client;
+mod ethereum_deploy_contract;
 mod ethereum_sync_loop;
 mod ethereum_types;
 mod headers;
 mod substrate_client;
+mod substrate_sync_loop;
 mod substrate_types;
 mod sync;
 mod sync_loop;
 mod sync_types;
+mod utils;
 
+use ethereum_client::{EthereumConnectionParams, EthereumSigningParams};
+use ethereum_sync_loop::EthereumSyncParams;
+use parity_crypto::publickey::{KeyPair, Secret};
 use sp_core::crypto::Pair;
 use std::io::Write;
+use substrate_client::{SubstrateConnectionParams, SubstrateSigningParams};
+use substrate_sync_loop::SubstrateSyncParams;
 
 fn main() {
 	initialize();
@@ -44,6 +52,24 @@ fn main() {
 				}
 			});
 		}
+		("sub-to-eth", Some(sub_to_eth_matches)) => {
+			substrate_sync_loop::run(match substrate_sync_params(&sub_to_eth_matches) {
+				Ok(substrate_sync_params) => substrate_sync_params,
+				Err(err) => {
+					log::error!(target: "bridge", "Error parsing parameters: {}", err);
+					return;
+				}
+			});
+		}
+		("eth-deploy-contract", Some(eth_deploy_matches)) => {
+			ethereum_deploy_contract::run(match ethereum_deploy_contract_params(&eth_deploy_matches) {
+				Ok(ethereum_deploy_matches) => ethereum_deploy_matches,
+				Err(err) => {
+					log::error!(target: "bridge", "Error parsing parameters: {}", err);
+					return;
+				}
+			});
+		}
 		("", _) => {
 			log::error!(target: "bridge", "No subcommand specified");
 			return;
@@ -89,25 +115,58 @@ fn initialize() {
 	builder.init();
 }
 
-fn ethereum_sync_params(matches: &clap::ArgMatches) -> Result<ethereum_sync_loop::EthereumSyncParams, String> {
+fn ethereum_connection_params(matches: &clap::ArgMatches) -> Result<EthereumConnectionParams, String> {
-	let mut eth_sync_params = ethereum_sync_loop::EthereumSyncParams::default();
+	let mut params = EthereumConnectionParams::default();
 	if let Some(eth_host) = matches.value_of("eth-host") {
-		eth_sync_params.eth_host = eth_host.into();
+		params.host = eth_host.into();
 	}
 	if let Some(eth_port) = matches.value_of("eth-port") {
-		eth_sync_params.eth_port = eth_port.parse().map_err(|e| format!("{}", e))?;
+		params.port = eth_port
+			.parse()
+			.map_err(|e| format!("Failed to parse eth-port: {}", e))?;
 	}
+	Ok(params)
+}
+
+fn ethereum_signing_params(matches: &clap::ArgMatches) -> Result<EthereumSigningParams, String> {
+	let mut params = EthereumSigningParams::default();
+	if let Some(eth_signer) = matches.value_of("eth-signer") {
+		params.signer = eth_signer
+			.parse::<Secret>()
+			.map_err(|e| format!("Failed to parse eth-signer: {}", e))
+			.and_then(|secret| KeyPair::from_secret(secret).map_err(|e| format!("Invalid eth-signer: {}", e)))?;
+	}
+	Ok(params)
+}
+
+fn substrate_connection_params(matches: &clap::ArgMatches) -> Result<SubstrateConnectionParams, String> {
+	let mut params = SubstrateConnectionParams::default();
 	if let Some(sub_host) = matches.value_of("sub-host") {
-		eth_sync_params.sub_host = sub_host.into();
+		params.host = sub_host.into();
 	}
 	if let Some(sub_port) = matches.value_of("sub-port") {
-		eth_sync_params.sub_port = sub_port.parse().map_err(|e| format!("{}", e))?;
+		params.port = sub_port
+			.parse()
+			.map_err(|e| format!("Failed to parse sub-port: {}", e))?;
 	}
+	Ok(params)
+}
+
+fn substrate_signing_params(matches: &clap::ArgMatches) -> Result<SubstrateSigningParams, String> {
+	let mut params = SubstrateSigningParams::default();
 	if let Some(sub_signer) = matches.value_of("sub-signer") {
 		let sub_signer_password = matches.value_of("sub-signer-password");
-		eth_sync_params.sub_signer =
+		params.signer = sp_core::sr25519::Pair::from_string(sub_signer, sub_signer_password)
-			sp_core::sr25519::Pair::from_string(sub_signer, sub_signer_password).map_err(|e| format!("{:?}", e))?;
+			.map_err(|e| format!("Failed to parse sub-signer: {:?}", e))?;
 	}
+	Ok(params)
+}
+
+fn ethereum_sync_params(matches: &clap::ArgMatches) -> Result<EthereumSyncParams, String> {
+	let mut eth_sync_params = EthereumSyncParams::default();
+	eth_sync_params.eth = ethereum_connection_params(matches)?;
+	eth_sync_params.sub = substrate_connection_params(matches)?;
+	eth_sync_params.sub_sign = substrate_signing_params(matches)?;
 
 	match matches.value_of("sub-tx-mode") {
 		Some("signed") => eth_sync_params.sync_params.target_tx_mode = sync::TargetTransactionMode::Signed,
@@ -124,3 +183,32 @@ fn ethereum_sync_params(matches: &clap::ArgMatches) -> Result<ethereum_sync_loop
 
 	Ok(eth_sync_params)
 }
+
+fn substrate_sync_params(matches: &clap::ArgMatches) -> Result<SubstrateSyncParams, String> {
+	let mut sub_sync_params = SubstrateSyncParams::default();
+	sub_sync_params.eth = ethereum_connection_params(matches)?;
+	sub_sync_params.eth_sign = ethereum_signing_params(matches)?;
+	sub_sync_params.sub = substrate_connection_params(matches)?;
+
+	if let Some(eth_contract) = matches.value_of("eth-contract") {
+		sub_sync_params.eth_contract_address = eth_contract.parse().map_err(|e| format!("{}", e))?;
+	}
+
+	Ok(sub_sync_params)
+}
+
+fn ethereum_deploy_contract_params(
+	matches: &clap::ArgMatches,
+) -> Result<ethereum_deploy_contract::EthereumDeployContractParams, String> {
+	let mut eth_deploy_params = ethereum_deploy_contract::EthereumDeployContractParams::default();
+	eth_deploy_params.eth = ethereum_connection_params(matches)?;
+	eth_deploy_params.eth_sign = ethereum_signing_params(matches)?;
+	eth_deploy_params.sub = substrate_connection_params(matches)?;
+
+	if let Some(eth_contract_code) = matches.value_of("eth-contract-code") {
+		eth_deploy_params.eth_contract_code =
+			hex::decode(&eth_contract_code).map_err(|e| format!("Failed to parse eth-contract-code: {}", e))?;
+	}
+
+	Ok(eth_deploy_params)
+}

bridges/relays/ethereum/src/substrate_client.rs

@@ -15,16 +15,60 @@
 // along with Parity Bridges Common.  If not, see <http://www.gnu.org/licenses/>.
 
 use crate::ethereum_types::{Bytes, EthereumHeaderId, QueuedEthereumHeader, H256};
-use crate::substrate_types::{into_substrate_ethereum_header, into_substrate_ethereum_receipts, TransactionHash};
+use crate::substrate_types::{
+	into_substrate_ethereum_header, into_substrate_ethereum_receipts, Hash, Header as SubstrateHeader, Number,
+};
 use crate::sync_types::{HeaderId, MaybeConnectionError, SourceHeader};
+use crate::{bail_on_arg_error, bail_on_error};
 use codec::{Decode, Encode};
 use jsonrpsee::common::Params;
 use jsonrpsee::raw::{RawClient, RawClientError};
 use jsonrpsee::transport::http::{HttpTransportClient, RequestError};
-use serde_json::{from_value, to_value};
+use num_traits::Zero;
+use serde::de::DeserializeOwned;
+use serde_json::{from_value, to_value, Value};
 use sp_core::crypto::Pair;
 use sp_runtime::traits::IdentifyAccount;
 
+/// Substrate connection params.
+#[derive(Debug)]
+pub struct SubstrateConnectionParams {
+	/// Substrate RPC host.
+	pub host: String,
+	/// Substrate RPC port.
+	pub port: u16,
+}
+
+impl Default for SubstrateConnectionParams {
+	fn default() -> Self {
+		SubstrateConnectionParams {
+			host: "localhost".into(),
+			port: 9933,
+		}
+	}
+}
+
+/// Substrate signing params.
+#[derive(Clone)]
+pub struct SubstrateSigningParams {
+	/// Substrate transactions signer.
+	pub signer: sp_core::sr25519::Pair,
+}
+
+impl std::fmt::Debug for SubstrateSigningParams {
+	fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
+		write!(f, "{}", self.signer.public())
+	}
+}
+
+impl Default for SubstrateSigningParams {
+	fn default() -> Self {
+		SubstrateSigningParams {
+			signer: sp_keyring::AccountKeyring::Alice.pair(),
+		}
+	}
+}
+
 /// Substrate client type.
 pub struct Client {
 	/// Substrate RPC client.
@@ -38,6 +82,8 @@ pub struct Client {
 pub enum Error {
 	/// Request start failed.
 	StartRequestFailed(RequestError),
+	/// Error serializing request.
+	RequestSerialization(serde_json::Error),
 	/// Request not found (should never occur?).
 	RequestNotFound,
 	/// Failed to receive response.
@@ -56,23 +102,42 @@ impl MaybeConnectionError for Error {
 }
 
 /// Returns client that is able to call RPCs on Substrate node.
-pub fn client(uri: &str) -> Client {
+pub fn client(params: SubstrateConnectionParams) -> Client {
-	let transport = HttpTransportClient::new(uri);
+	let uri = format!("http://{}:{}", params.host, params.port);
+	let transport = HttpTransportClient::new(&uri);
 	Client {
 		rpc_client: RawClient::new(transport),
 		genesis_hash: None,
 	}
 }
 
+/// Returns best Substrate header.
+pub async fn best_header(client: Client) -> (Client, Result<SubstrateHeader, Error>) {
+	call_rpc(client, "chain_getHeader", Params::None, rpc_returns_value).await
+}
+
+/// Returns Substrate header by hash.
+pub async fn header_by_hash(client: Client, hash: Hash) -> (Client, Result<SubstrateHeader, Error>) {
+	let hash = bail_on_arg_error!(to_value(hash).map_err(|e| Error::RequestSerialization(e)), client);
+	call_rpc(client, "chain_getHeader", Params::Array(vec![hash]), rpc_returns_value).await
+}
+
+/// Returns Substrate header by number.
+pub async fn header_by_number(client: Client, number: Number) -> (Client, Result<SubstrateHeader, Error>) {
+	let (client, hash) = bail_on_error!(block_hash_by_number(client, number).await);
+	header_by_hash(client, hash).await
+}
+
 /// Returns best Ethereum block that Substrate runtime knows of.
 pub async fn best_ethereum_block(client: Client) -> (Client, Result<EthereumHeaderId, Error>) {
-	let (client, result) = call_rpc::<(u64, H256)>(
+	let (client, result) = call_rpc(
 		client,
 		"state_call",
 		Params::Array(vec![
-			to_value("EthereumHeadersApi_best_block").unwrap(),
+			serde_json::Value::String("EthereumHeadersApi_best_block".into()),
-			to_value("0x").unwrap(),
+			serde_json::Value::String("0x".into()),
 		]),
+		rpc_returns_encoded_value,
 	)
 	.await;
 	(client, result.map(|(num, hash)| HeaderId(num, hash)))
@@ -85,14 +150,18 @@ pub async fn ethereum_receipts_required(
 ) -> (Client, Result<(EthereumHeaderId, bool), Error>) {
 	let id = header.header().id();
 	let header = into_substrate_ethereum_header(header.header());
-	let encoded_header = header.encode();
+	let encoded_header = bail_on_arg_error!(
+		to_value(Bytes(header.encode())).map_err(|e| Error::RequestSerialization(e)),
+		client
+	);
 	let (client, receipts_required) = call_rpc(
 		client,
 		"state_call",
 		Params::Array(vec![
-			to_value("EthereumHeadersApi_is_import_requires_receipts").unwrap(),
+			serde_json::Value::String("EthereumHeadersApi_is_import_requires_receipts".into()),
-			to_value(Bytes(encoded_header)).unwrap(),
+			encoded_header,
 		]),
+		rpc_returns_encoded_value,
 	)
 	.await;
 	(
@@ -112,14 +181,18 @@ pub async fn ethereum_header_known(
 	// But when we'll read best header from Substrate next time, we will know that
 	// there's a better header => this Orphan will either be marked as synced, or
 	// eventually pruned.
-	let encoded_id = id.1.encode();
+	let encoded_id = bail_on_arg_error!(
+		to_value(Bytes(id.1.encode())).map_err(|e| Error::RequestSerialization(e)),
+		client
+	);
 	let (client, is_known_block) = call_rpc(
 		client,
 		"state_call",
 		Params::Array(vec![
-			to_value("EthereumHeadersApi_is_known_block").unwrap(),
+			serde_json::Value::String("EthereumHeadersApi_is_known_block".into()),
-			to_value(Bytes(encoded_id)).unwrap(),
+			encoded_id,
 		]),
+		rpc_returns_encoded_value,
 	)
 	.await;
 	(client, is_known_block.map(|is_known_block| (id, is_known_block)))
@@ -128,12 +201,12 @@ pub async fn ethereum_header_known(
 /// Submits Ethereum header to Substrate runtime.
 pub async fn submit_ethereum_headers(
 	client: Client,
-	signer: sp_core::sr25519::Pair,
+	params: SubstrateSigningParams,
 	headers: Vec<QueuedEthereumHeader>,
 	sign_transactions: bool,
-) -> (Client, Result<(Vec<TransactionHash>, Vec<EthereumHeaderId>), Error>) {
+) -> (Client, Result<Vec<EthereumHeaderId>, Error>) {
 	match sign_transactions {
-		true => submit_signed_ethereum_headers(client, signer, headers).await,
+		true => submit_signed_ethereum_headers(client, params, headers).await,
 		false => submit_unsigned_ethereum_headers(client, headers).await,
 	}
 }
@@ -141,78 +214,89 @@ pub async fn submit_ethereum_headers(
 /// Submits signed Ethereum header to Substrate runtime.
 pub async fn submit_signed_ethereum_headers(
 	client: Client,
-	signer: sp_core::sr25519::Pair,
+	params: SubstrateSigningParams,
 	headers: Vec<QueuedEthereumHeader>,
-) -> (Client, Result<(Vec<TransactionHash>, Vec<EthereumHeaderId>), Error>) {
+) -> (Client, Result<Vec<EthereumHeaderId>, Error>) {
 	let ids = headers.iter().map(|header| header.id()).collect();
 	let (client, genesis_hash) = match client.genesis_hash {
 		Some(genesis_hash) => (client, genesis_hash),
 		None => {
-			let (mut client, genesis_hash) = block_hash_by_number(client, 0).await;
+			let (mut client, genesis_hash) = bail_on_error!(block_hash_by_number(client, Zero::zero()).await);
-			let genesis_hash = match genesis_hash {
-				Ok(genesis_hash) => genesis_hash,
-				Err(err) => return (client, Err(err)),
-			};
 			client.genesis_hash = Some(genesis_hash);
 			(client, genesis_hash)
 		}
 	};
-	let account_id = signer.public().as_array_ref().clone().into();
+	let account_id = params.signer.public().as_array_ref().clone().into();
-	let (client, nonce) = next_account_index(client, account_id).await;
+	let (client, nonce) = bail_on_error!(next_account_index(client, account_id).await);
-	let nonce = match nonce {
-		Ok(nonce) => nonce,
-		Err(err) => return (client, Err(err)),
-	};
 
-	let transaction = create_signed_submit_transaction(headers, &signer, nonce, genesis_hash);
+	let transaction = create_signed_submit_transaction(headers, &params.signer, nonce, genesis_hash);
-	let encoded_transaction = transaction.encode();
+	let encoded_transaction = bail_on_arg_error!(
-	let (client, transaction_hash) = call_rpc(
+		to_value(Bytes(transaction.encode())).map_err(|e| Error::RequestSerialization(e)),
+		client
+	);
+	let (client, _) = call_rpc(
 		client,
 		"author_submitExtrinsic",
-		Params::Array(vec![to_value(Bytes(encoded_transaction)).unwrap()]),
+		Params::Array(vec![encoded_transaction]),
+		|_| Ok(()),
 	)
 	.await;
 
-	(
+	(client, Ok(ids))
-		client,
-		transaction_hash.map(|transaction_hash| (vec![transaction_hash], ids)),
-	)
 }
 
 /// Submits unsigned Ethereum header to Substrate runtime.
 pub async fn submit_unsigned_ethereum_headers(
 	mut client: Client,
 	headers: Vec<QueuedEthereumHeader>,
-) -> (Client, Result<(Vec<TransactionHash>, Vec<EthereumHeaderId>), Error>) {
+) -> (Client, Result<Vec<EthereumHeaderId>, Error>) {
 	let ids = headers.iter().map(|header| header.id()).collect();
-	let mut transactions_hashes = Vec::new();
 	for header in headers {
 		let transaction = create_unsigned_submit_transaction(header);
 
-		let encoded_transaction = transaction.encode();
+		let encoded_transaction = bail_on_arg_error!(
-		let (used_client, transaction_hash) = call_rpc(
+			to_value(Bytes(transaction.encode())).map_err(|e| Error::RequestSerialization(e)),
-			client,
+			client
-			"author_submitExtrinsic",
+		);
-			Params::Array(vec![to_value(Bytes(encoded_transaction)).unwrap()]),
+		let (used_client, _) = bail_on_error!(
-		)
+			call_rpc(
-		.await;
+				client,
+				"author_submitExtrinsic",
+				Params::Array(vec![encoded_transaction]),
+				|_| Ok(()),
+			)
+			.await
+		);
 
 		client = used_client;
-		transactions_hashes.push(match transaction_hash {
-			Ok(transaction_hash) => transaction_hash,
-			Err(error) => return (client, Err(error)),
-		});
 	}
 
-	(client, Ok((transactions_hashes, ids)))
+	(client, Ok(ids))
+}
+
+/// Get GRANDPA authorities set at given block.
+pub async fn grandpa_authorities_set(client: Client, block: Hash) -> (Client, Result<Vec<u8>, Error>) {
+	let block = bail_on_arg_error!(to_value(block).map_err(|e| Error::RequestSerialization(e)), client);
+	call_rpc(
+		client,
+		"state_call",
+		Params::Array(vec![
+			serde_json::Value::String("GrandpaApi_grandpa_authorities".into()),
+			block,
+		]),
+		rpc_returns_bytes,
+	)
+	.await
 }
 
 /// Get Substrate block hash by its number.
-async fn block_hash_by_number(client: Client, number: u64) -> (Client, Result<H256, Error>) {
+async fn block_hash_by_number(client: Client, number: Number) -> (Client, Result<Hash, Error>) {
+	let number = bail_on_arg_error!(to_value(number).map_err(|e| Error::RequestSerialization(e)), client);
 	call_rpc(
 		client,
 		"chain_getBlockHash",
-		Params::Array(vec![to_value(number).unwrap()]),
+		Params::Array(vec![number]),
+		rpc_returns_value,
 	)
 	.await
 }
@@ -224,18 +308,30 @@ async fn next_account_index(
 ) -> (Client, Result<node_primitives::Index, Error>) {
 	use sp_core::crypto::Ss58Codec;
 
-	let (client, index) = call_rpc_u64(
+	let account = bail_on_arg_error!(
-		client,
+		to_value(account.to_ss58check()).map_err(|e| Error::RequestSerialization(e)),
-		"system_accountNextIndex",
+		client
-		Params::Array(vec![to_value(account.to_ss58check()).unwrap()]),
+	);
-	)
+	let (client, index) = call_rpc(client, "system_accountNextIndex", Params::Array(vec![account]), |v| {
+		rpc_returns_value::<u64>(v)
+	})
 	.await;
 	(client, index.map(|index| index as _))
 }
 
 /// Calls RPC on Substrate node that returns Bytes.
-async fn call_rpc<T: Decode>(mut client: Client, method: &'static str, params: Params) -> (Client, Result<T, Error>) {
+async fn call_rpc<T>(
-	async fn do_call_rpc<T: Decode>(client: &mut Client, method: &'static str, params: Params) -> Result<T, Error> {
+	mut client: Client,
+	method: &'static str,
+	params: Params,
+	decode_value: impl Fn(Value) -> Result<T, Error>,
+) -> (Client, Result<T, Error>) {
+	async fn do_call_rpc<T>(
+		client: &mut Client,
+		method: &'static str,
+		params: Params,
+		decode_value: impl Fn(Value) -> Result<T, Error>,
+	) -> Result<T, Error> {
 		let request_id = client
 			.rpc_client
 			.start_request(method, params)
@@ -249,34 +345,10 @@ async fn call_rpc<T: Decode>(mut client: Client, method: &'static str, params: P
 			.ok_or(Error::RequestNotFound)?
 			.await
 			.map_err(Error::ResponseRetrievalFailed)?;
-		let encoded_response: Bytes = from_value(response).map_err(|_| Error::ResponseParseFailed)?;
+		decode_value(response)
-		Decode::decode(&mut &encoded_response.0[..]).map_err(|_| Error::ResponseParseFailed)
 	}
 
-	let result = do_call_rpc(&mut client, method, params).await;
+	let result = do_call_rpc(&mut client, method, params, decode_value).await;
-	(client, result)
-}
-
-/// Calls RPC on Substrate node that returns u64.
-async fn call_rpc_u64(mut client: Client, method: &'static str, params: Params) -> (Client, Result<u64, Error>) {
-	async fn do_call_rpc(client: &mut Client, method: &'static str, params: Params) -> Result<u64, Error> {
-		let request_id = client
-			.rpc_client
-			.start_request(method, params)
-			.await
-			.map_err(Error::StartRequestFailed)?;
-		// WARN: if there'll be need for executing >1 request at a time, we should avoid
-		// calling request_by_id
-		let response = client
-			.rpc_client
-			.request_by_id(request_id)
-			.ok_or(Error::RequestNotFound)?
-			.await
-			.map_err(Error::ResponseRetrievalFailed)?;
-		response.as_u64().ok_or(Error::ResponseParseFailed)
-	}
-
-	let result = do_call_rpc(&mut client, method, params).await;
 	(client, result)
 }
 
@@ -341,3 +413,20 @@ fn create_unsigned_submit_transaction(header: QueuedEthereumHeader) -> bridge_no
 
 	bridge_node_runtime::UncheckedExtrinsic::new_unsigned(function)
 }
+
+/// When RPC method returns encoded value.
+fn rpc_returns_encoded_value<T: Decode>(value: Value) -> Result<T, Error> {
+	let encoded_response: Bytes = from_value(value).map_err(|_| Error::ResponseParseFailed)?;
+	Decode::decode(&mut &encoded_response.0[..]).map_err(|_| Error::ResponseParseFailed)
+}
+
+/// When RPC method returns value.
+fn rpc_returns_value<T: DeserializeOwned>(value: Value) -> Result<T, Error> {
+	from_value(value).map_err(|_| Error::ResponseParseFailed)
+}
+
+/// When RPC method returns raw bytes.
+fn rpc_returns_bytes(value: Value) -> Result<Vec<u8>, Error> {
+	let encoded_response: Bytes = from_value(value).map_err(|_| Error::ResponseParseFailed)?;
+	Ok(encoded_response.0)
+}

bridges/relays/ethereum/src/substrate_sync_loop.rs

@@ -0,0 +1,202 @@
+// Copyright 2019-2020 Parity Technologies (UK) Ltd.
+// This file is part of Parity Bridges Common.
+
+// Parity Bridges Common 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 Bridges Common 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 Bridges Common.  If not, see <http://www.gnu.org/licenses/>.
+
+use crate::ethereum_client::{self, EthereumConnectionParams, EthereumSigningParams};
+use crate::ethereum_types::Address;
+use crate::substrate_client::{self, SubstrateConnectionParams};
+use crate::substrate_types::{
+	Hash, Header, Number, QueuedSubstrateHeader, SubstrateHeaderId, SubstrateHeadersSyncPipeline,
+};
+use crate::sync::{HeadersSyncParams, TargetTransactionMode};
+use crate::sync_loop::{SourceClient, TargetClient};
+use crate::sync_types::SourceHeader;
+use futures::future::{ready, FutureExt, Ready};
+use std::{future::Future, pin::Pin};
+
+/// Interval (in ms) at which we check new Substrate headers when we are synced/almost synced.
+const SUBSTRATE_TICK_INTERVAL_MS: u64 = 10_000;
+/// Interval (in ms) at which we check new Ethereum blocks.
+const ETHEREUM_TICK_INTERVAL_MS: u64 = 5_000;
+/// Max Ethereum headers we want to have in all 'before-submitted' states.
+const MAX_FUTURE_HEADERS_TO_DOWNLOAD: usize = 8;
+/// Max Ethereum headers count we want to have in 'submitted' state.
+const MAX_SUBMITTED_HEADERS: usize = 4;
+/// Max depth of in-memory headers in all states. Past this depth they will be forgotten (pruned).
+const PRUNE_DEPTH: u32 = 256;
+
+/// Substrate synchronization parameters.
+#[derive(Debug)]
+pub struct SubstrateSyncParams {
+	/// Ethereum connection params.
+	pub eth: EthereumConnectionParams,
+	/// Ethereum signing params.
+	pub eth_sign: EthereumSigningParams,
+	/// Ethereum bridge contract address.
+	pub eth_contract_address: Address,
+	/// Substrate connection params.
+	pub sub: SubstrateConnectionParams,
+	/// Synchronization parameters.
+	pub sync_params: HeadersSyncParams,
+}
+
+impl Default for SubstrateSyncParams {
+	fn default() -> Self {
+		SubstrateSyncParams {
+			eth: Default::default(),
+			eth_sign: Default::default(),
+			// the address 0x731a10897d267e19b34503ad902d0a29173ba4b1 is the address
+			// of the contract that is deployed by default signer and 0 nonce
+			eth_contract_address: "731a10897d267e19b34503ad902d0a29173ba4b1"
+				.parse()
+				.expect("address is hardcoded, thus valid; qed"),
+			sub: Default::default(),
+			sync_params: HeadersSyncParams {
+				max_future_headers_to_download: MAX_FUTURE_HEADERS_TO_DOWNLOAD,
+				max_headers_in_submitted_status: MAX_SUBMITTED_HEADERS,
+				// since we always have single Substrate header in separate Ethereum transaction,
+				// all max_**_in_single_submit aren't important here
+				max_headers_in_single_submit: 4,
+				max_headers_size_in_single_submit: std::usize::MAX,
+				prune_depth: PRUNE_DEPTH,
+				target_tx_mode: TargetTransactionMode::Signed,
+			},
+		}
+	}
+}
+
+/// Substrate client as headers source.
+struct SubstrateHeadersSource {
+	/// Substrate node client.
+	client: substrate_client::Client,
+}
+
+impl SourceClient<SubstrateHeadersSyncPipeline> for SubstrateHeadersSource {
+	type Error = substrate_client::Error;
+	type BestBlockNumberFuture = Pin<Box<dyn Future<Output = (Self, Result<Number, Self::Error>)>>>;
+	type HeaderByHashFuture = Pin<Box<dyn Future<Output = (Self, Result<Header, Self::Error>)>>>;
+	type HeaderByNumberFuture = Pin<Box<dyn Future<Output = (Self, Result<Header, Self::Error>)>>>;
+	type HeaderExtraFuture = Ready<(Self, Result<(SubstrateHeaderId, ()), Self::Error>)>;
+
+	fn best_block_number(self) -> Self::BestBlockNumberFuture {
+		substrate_client::best_header(self.client)
+			.map(|(client, result)| (SubstrateHeadersSource { client }, result.map(|header| header.number)))
+			.boxed()
+	}
+
+	fn header_by_hash(self, hash: Hash) -> Self::HeaderByHashFuture {
+		substrate_client::header_by_hash(self.client, hash)
+			.map(|(client, result)| (SubstrateHeadersSource { client }, result))
+			.boxed()
+	}
+
+	fn header_by_number(self, number: Number) -> Self::HeaderByNumberFuture {
+		substrate_client::header_by_number(self.client, number)
+			.map(|(client, result)| (SubstrateHeadersSource { client }, result))
+			.boxed()
+	}
+
+	fn header_extra(self, id: SubstrateHeaderId, _header: &Header) -> Self::HeaderExtraFuture {
+		ready((self, Ok((id, ()))))
+	}
+}
+
+/// Ethereum client as Substrate headers target.
+struct EthereumHeadersTarget {
+	/// Ethereum node client.
+	client: ethereum_client::Client,
+	/// Bridge contract address.
+	contract: Address,
+	/// Ethereum signing params.
+	sign_params: EthereumSigningParams,
+}
+
+impl TargetClient<SubstrateHeadersSyncPipeline> for EthereumHeadersTarget {
+	type Error = ethereum_client::Error;
+	type BestHeaderIdFuture = Pin<Box<dyn Future<Output = (Self, Result<SubstrateHeaderId, Self::Error>)>>>;
+	type IsKnownHeaderFuture = Pin<Box<dyn Future<Output = (Self, Result<(SubstrateHeaderId, bool), Self::Error>)>>>;
+	type RequiresExtraFuture = Ready<(Self, Result<(SubstrateHeaderId, bool), Self::Error>)>;
+	type SubmitHeadersFuture = Pin<Box<dyn Future<Output = (Self, Result<Vec<SubstrateHeaderId>, Self::Error>)>>>;
+
+	fn best_header_id(self) -> Self::BestHeaderIdFuture {
+		let (contract, sign_params) = (self.contract, self.sign_params);
+		ethereum_client::best_substrate_block(self.client, contract)
+			.map(move |(client, result)| {
+				(
+					EthereumHeadersTarget {
+						client,
+						contract,
+						sign_params,
+					},
+					result,
+				)
+			})
+			.boxed()
+	}
+
+	fn is_known_header(self, id: SubstrateHeaderId) -> Self::IsKnownHeaderFuture {
+		let (contract, sign_params) = (self.contract, self.sign_params);
+		ethereum_client::substrate_header_known(self.client, contract, id)
+			.map(move |(client, result)| {
+				(
+					EthereumHeadersTarget {
+						client,
+						contract,
+						sign_params,
+					},
+					result,
+				)
+			})
+			.boxed()
+	}
+
+	fn requires_extra(self, header: &QueuedSubstrateHeader) -> Self::RequiresExtraFuture {
+		ready((self, Ok((header.header().id(), false))))
+	}
+
+	fn submit_headers(self, headers: Vec<QueuedSubstrateHeader>) -> Self::SubmitHeadersFuture {
+		let (contract, sign_params) = (self.contract, self.sign_params);
+		ethereum_client::submit_substrate_headers(self.client, sign_params.clone(), contract, headers)
+			.map(move |(client, result)| {
+				(
+					EthereumHeadersTarget {
+						client,
+						contract,
+						sign_params,
+					},
+					result,
+				)
+			})
+			.boxed()
+	}
+}
+
+/// Run Substrate headers synchronization.
+pub fn run(params: SubstrateSyncParams) {
+	let eth_client = ethereum_client::client(params.eth);
+	let sub_client = substrate_client::client(params.sub);
+
+	crate::sync_loop::run(
+		SubstrateHeadersSource { client: sub_client },
+		SUBSTRATE_TICK_INTERVAL_MS,
+		EthereumHeadersTarget {
+			client: eth_client,
+			contract: params.eth_contract_address,
+			sign_params: params.eth_sign,
+		},
+		ETHEREUM_TICK_INTERVAL_MS,
+		params.sync_params,
+	);
+}

bridges/relays/ethereum/src/substrate_types.rs

@@ -14,16 +14,61 @@
 // You should have received a copy of the GNU General Public License
 // along with Parity Bridges Common.  If not, see <http://www.gnu.org/licenses/>.
 
-pub use crate::ethereum_types::H256 as TransactionHash;
 use crate::ethereum_types::{
 	Header as EthereumHeader, Receipt as EthereumReceipt, HEADER_ID_PROOF as ETHEREUM_HEADER_ID_PROOF,
 	RECEIPT_GAS_USED_PROOF as ETHEREUM_RECEIPT_GAS_USED_PROOF,
 };
+use crate::sync_types::{HeaderId, HeadersSyncPipeline, QueuedHeader, SourceHeader};
+use codec::Encode;
 pub use sp_bridge_eth_poa::{
 	Address, Bloom, Bytes, Header as SubstrateEthereumHeader, LogEntry as SubstrateEthereumLogEntry,
 	Receipt as SubstrateEthereumReceipt, TransactionOutcome as SubstrateEthereumTransactionOutcome, H256, U256,
 };
 
+/// Substrate header hash.
+pub type Hash = bridge_node_runtime::Hash;
+
+/// Substrate header number.
+pub type Number = bridge_node_runtime::BlockNumber;
+
+/// Substrate header type.
+pub type Header = bridge_node_runtime::Header;
+
+/// Substrate header ID.
+pub type SubstrateHeaderId = HeaderId<bridge_node_runtime::Hash, bridge_node_runtime::BlockNumber>;
+
+/// Queued substrate header ID.
+pub type QueuedSubstrateHeader = QueuedHeader<SubstrateHeadersSyncPipeline>;
+
+/// Substrate synchronization pipeline.
+#[derive(Clone, Copy, Debug)]
+#[cfg_attr(test, derive(PartialEq))]
+pub struct SubstrateHeadersSyncPipeline;
+
+impl HeadersSyncPipeline for SubstrateHeadersSyncPipeline {
+	const SOURCE_NAME: &'static str = "Substrate";
+	const TARGET_NAME: &'static str = "Ethereum";
+
+	type Hash = bridge_node_runtime::Hash;
+	type Number = bridge_node_runtime::BlockNumber;
+	type Header = Header;
+	type Extra = ();
+
+	fn estimate_size(source: &QueuedHeader<Self>) -> usize {
+		source.header().encode().len()
+	}
+}
+
+impl SourceHeader<bridge_node_runtime::Hash, bridge_node_runtime::BlockNumber> for Header {
+	fn id(&self) -> SubstrateHeaderId {
+		HeaderId(self.number, self.hash())
+	}
+
+	fn parent_id(&self) -> SubstrateHeaderId {
+		HeaderId(self.number - 1, self.parent_hash)
+	}
+}
+
 /// Convert Ethereum header into Ethereum header for Substrate.
 pub fn into_substrate_ethereum_header(header: &EthereumHeader) -> SubstrateEthereumHeader {
 	SubstrateEthereumHeader {

bridges/relays/ethereum/src/sync.rs

@@ -204,19 +204,6 @@ impl<P: HeadersSyncPipeline> HeadersSync<P> {
 	}
 }
 
-impl Default for HeadersSyncParams {
-	fn default() -> Self {
-		HeadersSyncParams {
-			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,
-			target_tx_mode: TargetTransactionMode::Signed,
-		}
-	}
-}
-
 #[cfg(test)]
 mod tests {
 	use super::*;
@@ -228,9 +215,20 @@ mod tests {
 		H256::from_low_u64_le(1000 + number)
 	}
 
+	fn default_sync_params() -> HeadersSyncParams {
+		HeadersSyncParams {
+			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,
+			target_tx_mode: TargetTransactionMode::Signed,
+		}
+	}
+
 	#[test]
 	fn select_new_header_to_download_works() {
-		let mut eth_sync = HeadersSync::<EthereumHeadersSyncPipeline>::new(Default::default());
+		let mut eth_sync = HeadersSync::<EthereumHeadersSyncPipeline>::new(default_sync_params());
 
 		// both best && target headers are unknown
 		assert_eq!(eth_sync.select_new_header_to_download(), None);
@@ -261,7 +259,7 @@ mod tests {
 
 	#[test]
 	fn sync_without_reorgs_works() {
-		let mut eth_sync = HeadersSync::new(Default::default());
+		let mut eth_sync = HeadersSync::new(default_sync_params());
 		eth_sync.params.max_headers_in_submitted_status = 1;
 
 		// ethereum reports best header #102
@@ -310,7 +308,7 @@ mod tests {
 
 	#[test]
 	fn sync_with_orphan_headers_work() {
-		let mut eth_sync = HeadersSync::new(Default::default());
+		let mut eth_sync = HeadersSync::new(default_sync_params());
 
 		// ethereum reports best header #102
 		eth_sync.source_best_header_number_response(102);
@@ -359,7 +357,7 @@ mod tests {
 
 	#[test]
 	fn pruning_happens_on_target_best_header_response() {
-		let mut eth_sync = HeadersSync::<EthereumHeadersSyncPipeline>::new(Default::default());
+		let mut eth_sync = HeadersSync::<EthereumHeadersSyncPipeline>::new(default_sync_params());
 		eth_sync.params.prune_depth = 50;
 		eth_sync.target_best_header_response(id(100));
 		assert_eq!(eth_sync.headers.prune_border(), 50);
@@ -367,7 +365,7 @@ mod tests {
 
 	#[test]
 	fn only_submitting_headers_in_backup_mode_when_stalled() {
-		let mut eth_sync = HeadersSync::new(Default::default());
+		let mut eth_sync = HeadersSync::new(default_sync_params());
 		eth_sync.params.target_tx_mode = TargetTransactionMode::Backup;
 
 		// ethereum reports best header #102

bridges/relays/ethereum/src/sync_loop.rs

@@ -17,7 +17,7 @@
 use crate::sync::HeadersSyncParams;
 use crate::sync_types::{HeaderId, HeaderStatus, HeadersSyncPipeline, MaybeConnectionError, QueuedHeader};
 use futures::{future::FutureExt, stream::StreamExt};
-use num_traits::Saturating;
+use num_traits::{Saturating, Zero};
 use std::future::Future;
 
 /// When we submit headers to target node, but see no updates of best
@@ -365,6 +365,17 @@ pub fn run<P: HeadersSyncPipeline>(
 					// for Orphan we actually ask for parent' header
 					let parent_id = header.parent_id();
 
+					// if we have end up with orphan header#0, then we are misconfigured
+					if parent_id.0.is_zero() {
+						log::error!(
+							target: "bridge",
+							"Misconfiguration. Genesis {} header is considered orphan by {} node",
+							P::SOURCE_NAME,
+							P::TARGET_NAME,
+						);
+						return;
+					}
+
 					log::debug!(
 						target: "bridge",
 						"Going to download orphan header from {} node: {:?}",

bridges/relays/ethereum/src/utils.rs

@@ -0,0 +1,37 @@
+// Copyright 2019-2020 Parity Technologies (UK) Ltd.
+// This file is part of Parity Bridges Common.
+
+// Parity Bridges Common 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 Bridges Common 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 Bridges Common.  If not, see <http://www.gnu.org/licenses/>.
+
+/// Macro that returns (client, Err(error)) tuple from function if result is Err(error).
+#[macro_export]
+macro_rules! bail_on_error {
+	($result: expr) => {
+		match $result {
+			(client, Ok(result)) => (client, result),
+			(client, Err(error)) => return (client, Err(error)),
+			}
+	};
+}
+
+/// Macro that returns (client, Err(error)) tuple from function if result is Err(error).
+#[macro_export]
+macro_rules! bail_on_arg_error {
+	($result: expr, $client: ident) => {
+		match $result {
+			Ok(result) => result,
+			Err(error) => return ($client, Err(error)),
+			}
+	};
+}