update upstream

.env.example

@@ -5,6 +5,7 @@ ETH_VALIDATOR_ADDRESS="0xC6870AED119d01CE3f7A377775bD489839c51815"
 ETH_VALIDATOR_PRIVATE_KEY="0x0000000000000000000000000000000000000000000000000000000000000000"
 
 TOKEN_BRIDGE_ADDRESS="0x813a4658007ed3c7b31f02009e8699bef8539cd8"
+TOKEN_SYMBOL="DAI"
 SUB_TOKEN_INDEX="0"
 
 SUB_API_URL="ws://localhost:9944"

Cargo.toml

@@ -1,11 +1,9 @@
 [package]
-version = "0.2.1"
+version = "0.2.3"
 name = "validator"
 authors = ["Akropolis Decentralised Ltd., <team@akropolis.io>"]
 edition = "2018"
 
-# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
-
 [dependencies]
 web3 = "0.8.0"
 tokio = "0.1.22"
@@ -19,52 +17,24 @@ dotenv = "0.14.1"
 graphql_client = "0.8.0"
 reqwest = "0.9.22"
 serde = "1.0.101"
+serde_json = "1.0"
 time = "0.1.42"
 failure_derive = "0.1.7"
 
-
 raw-transaction-builder = { path = 'raw-transaction-builder' }
 
 substrate-bip39 = "0.3.1"
-node-runtime = { git = 'https://github.com/akropolisio/akropolisos-chain-node', version = '0.5.4', package = 'akropolisos-runtime', branch = 'v2.0.0-alpha.6' }
-substrate-api-client = { git = 'https://github.com/akropolisio/substrate-api-client', version = '2.0.2', branch = 'v2.0.0-alpha.6' }
-
-[dependencies.consensus]
-git = 'https://github.com/paritytech/substrate.git'
-package = 'pallet-babe'
-version = '2.0.0-alpha.6'
-
-[dependencies.system]
-git = 'https://github.com/paritytech/substrate.git'
-package = 'frame-system'
-version = '2.0.0-alpha.6'
-
-[dependencies.balances]
-default_features = false
-git = 'https://github.com/paritytech/substrate.git'
-package = 'pallet-balances'
-version = '2.0.0-alpha.6'
+node-runtime = { git = 'https://github.com/alekspickle/akropolisos-chain-node', version = '0.5.12', package = 'runtime', branch = 'aura' }
+substrate-api-client = { git = 'https://github.com/akropolisio/substrate-api-client', version = '0.4.5', branch = 'custom_node_alpha.7' }
 
 [dependencies.primitives]
 features = ["full_crypto"]
-git = 'https://github.com/paritytech/substrate.git'
 package = 'sp-core'
-version = '2.0.0-alpha.6'
-
-[dependencies.keyring]
-git = 'https://github.com/paritytech/substrate.git'
-version = '2.0.0-alpha.6'
-package = "sp-keyring"
+version = '2.0.0-alpha.7'
 
 [dependencies.runtime-primitives]
-git = 'https://github.com/paritytech/substrate.git'
 package = 'sp-runtime'
-version = '2.0.0-alpha.6'
-
-[dependencies.parity-codec]
-default-features = false
-features = ['derive']
-version = '3.5'
+version = '2.0.0-alpha.7'
 
 [dependencies.codec]
 package = "parity-scale-codec"

src/config.rs

@@ -16,6 +16,7 @@ pub struct Config {
     pub eth_validator_address: Address,
     pub eth_validator_private_key: String,
     pub token_bridge_address: Address,
+    pub token_symbol: String,
     pub eth_gas_price: u64,
     pub eth_gas: u64,
     pub sub_token_index: u32,
@@ -31,6 +32,7 @@ impl Config {
             eth_validator_address: parse_eth_validator_address()?,
             eth_validator_private_key: parse_eth_validator_private_key()?,
             token_bridge_address: parse_token_bridge_address()?,
+            token_symbol: parse_token_symbol()?,
             eth_gas_price: parse_eth_gas_price()?,
             eth_gas: parse_eth_gas()?,
             sub_token_index: parse_sub_token_index()?,
@@ -102,6 +104,13 @@ fn parse_sub_validator_mnemonic_phrase() -> Result<String, &'static str> {
     Ok(mnemonic_phrase)
 }
 
+fn parse_token_symbol() -> Result<String, &'static str> {
+    let sym = env::var("TOKEN_SYMBOL")
+        .map_err(|_| "can not read TOKEN_SYMBOL")?;
+
+    Ok(sym)
+}
+
 fn try_convert_to_sr25519_key_pair(mnemonic_phrase: &str) -> Result<(), &'static str> {
     sr25519::Pair::from_phrase(&mnemonic_phrase, None)
         .map_err(|_| "invalid SUB_VALIDATOR_MNEMONIC_PHRASE")?;

src/controller.rs

@@ -6,6 +6,7 @@ use std::thread;
 
 use crate::config::Config;
 use crate::controller_storage::ControllerStorage;
+use node_runtime::Balance;
 
 type MessageId = H256;
 type EthAddress = H160;
@@ -28,20 +29,8 @@ pub enum Event {
     EthBridgeStartedMessage(MessageId, EthAddress, BlockNumber),
     EthBridgeStoppedMessage(MessageId, EthAddress, BlockNumber),
 
-    EthRelayMessage(
-        MessageId,
-        EthAddress,
-        SubAddress,
-        Amount,
-        BlockNumber,
-    ),
-    EthApprovedRelayMessage(
-        MessageId,
-        EthAddress,
-        SubAddress,
-        Amount,
-        BlockNumber,
-    ),
+    EthRelayMessage(MessageId, EthAddress, SubAddress, Amount, BlockNumber),
+    EthApprovedRelayMessage(MessageId, EthAddress, SubAddress, Amount, BlockNumber),
     EthRevertMessage(MessageId, EthAddress, Amount, BlockNumber),
     EthWithdrawMessage(MessageId, BlockNumber),
 
@@ -89,6 +78,7 @@ pub enum Event {
 
     SubAccountPausedMessage(MessageId, SubAddress, Timestamp, TokenId, BlockNumber),
     SubAccountResumedMessage(MessageId, SubAddress, Timestamp, TokenId, BlockNumber),
+    OracleMessage(MessageId, Vec<u8>, Balance),
 }
 
 #[derive(Debug, PartialEq, Eq)]
@@ -99,7 +89,6 @@ enum EventType {
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 enum Status {
-    NotReady,
     Active,
     Paused,
     Stopped,
@@ -155,6 +144,7 @@ impl Event {
             Self::EthGuestAccountResumedMessage(message_id, _, _, _) => message_id,
             Self::SubAccountPausedMessage(message_id, _, _, _, _) => message_id,
             Self::SubAccountResumedMessage(message_id, _, _, _, _) => message_id,
+            Self::OracleMessage(message_id, _, _) => message_id,
         }
     }
 
@@ -184,6 +174,7 @@ impl Event {
             Self::EthGuestAccountResumedMessage(_, _, _, block_number) => *block_number,
             Self::SubAccountPausedMessage(_, _, _, _, block_number) => *block_number,
             Self::SubAccountResumedMessage(_, _, _, _, block_number) => *block_number,
+            Self::OracleMessage(_, _, _) => u128::default(),
         }
     }
 
@@ -249,24 +240,22 @@ impl Controller {
                         Status::Active => {
                             handle_account_control_events(storage, &event);
                             let deferred_events =
-                            storage.iter_events_queue().cloned().collect::<Vec<_>>();
+                                storage.iter_events_queue().cloned().collect::<Vec<_>>();
                             deferred_events.iter().cloned().for_each(|event| {
                                 handle_account_control_events(storage, &event);
                                 executor_tx.send(event).expect("can not sent event")
                             });
                             storage.clear_events_queue();
                             if event.event_type() == EventType::Transfer
-                            && storage.is_account_blocked(event.sender())
+                                && storage.is_account_blocked(event.sender())
                             {
                                 log::info!("putting event in a queue: {:?}", event);
                                 storage.put_event_to_account_queue(event)
                             } else {
                                 executor_tx.send(event).expect("can not sent event")
                             }
                         }
-                        Status::NotReady | Status::Paused | Status::Stopped => {
-                            storage.put_event_to_queue(event)
-                        }
+                        Status::Paused | Status::Stopped => storage.put_event_to_queue(event),
                     }
                 }
                 Err(e) => log::debug!("controller storage error: {:?}", e),
@@ -288,7 +277,7 @@ fn change_status(status: &mut Status, event: &Event) {
             }
             _ => (),
         },
-        Status::NotReady | Status::Paused => match event {
+        Status::Paused => match event {
             Event::EthBridgeResumedMessage(..) | Event::EthBridgeStartedMessage(..) => {
                 *status = Status::Active;
                 status_changed = true;

src/executor.rs

@@ -1,23 +1,20 @@
 use futures::future::{lazy, poll_fn};
 use log;
-use primitives::{self, crypto::Public};
+use primitives::{self, crypto::AccountId32};
 use tokio::runtime::{Runtime, TaskExecutor};
 use tokio_threadpool::blocking;
 use web3::{
     futures::Future,
     types::{Bytes, H160, H256, U256},
 };
 
+use crate::{config::Config, controller::Event, ethereum_transactions, substrate_transactions};
+use node_runtime::Balance;
 use std::{
     sync::{mpsc::Receiver, Arc},
     thread,
 };
 
-use crate::config::Config;
-use crate::controller::Event;
-use crate::ethereum_transactions;
-use crate::substrate_transactions;
-
 const AMOUNT: u64 = 0;
 
 #[derive(Debug)]
@@ -226,6 +223,9 @@ impl Executor {
                     message_id,
                     sub_address,
                 ),
+                Event::OracleMessage(_message_id, token, price) => {
+                    handle_oracle_message(&self.config, runtime.executor(), token, price)
+                }
             }
         })
     }
@@ -295,22 +295,25 @@ fn handle_eth_relay_message<T>(
 ) where
     T: web3::Transport + Send + Sync + 'static,
     T::Out: Send,
-    {
-    log::debug!("handle_eth_relay_message");
-
+{
     let args = (message_id, eth_address, sub_address, amount);
     let eth_validator_private_key = config.eth_validator_private_key.clone();
     let bridge_address = config.token_bridge_address;
     let eth_gas_price = config.eth_gas_price;
     let eth_gas = config.eth_gas;
 
-    log::debug!("handle_eth_relay_message:args:message_id:{:?} eth_address:{:?}, sub_address:{:?}, amount:{:?}", message_id, eth_address, sub_address, amount);
-
+    log::info!(
+        "handle_eth_relay_message: message_id:{:?} eth_address:{:?}, sub_address:{:?}, amount:{:?}",
+        message_id,
+        eth_address,
+        sub_address,
+        amount
+    );
+
     let data = ethereum_transactions::build_transaction_data(&abi, "approveTransfer", args);
     let fut = web3.eth().transaction_count(config.eth_validator_address, None)
         .and_then(move |nonce| {
 
-            log::debug!("approveTransfer input: bridge_address:{:?}, nonce:{:?}, AMOUNT:{:?}, eth gas price:{:?}, gas:{:?}, data:{:?}", bridge_address, nonce, AMOUNT, eth_gas_price, eth_gas, data);
             let tx = ethereum_transactions::build(eth_validator_private_key, bridge_address, nonce, AMOUNT, eth_gas_price, eth_gas, data);
             log::debug!("raw approveTransfer: {:?}", tx);
             web3.eth().send_raw_transaction(Bytes::from(tx))
@@ -341,15 +344,14 @@ fn handle_eth_approved_relay_message(
     sub_address: H256,
     amount: U256,
 ) {
-
     let message_id = primitives::H256::from_slice(&message_id.to_fixed_bytes());
     let eth_address = primitives::H160::from_slice(&eth_address.to_fixed_bytes());
     let sub_address = primitives::crypto::AccountId32::from(sub_address.to_fixed_bytes());
     let token_id = config.sub_token_index;
     let amount = amount.low_u128();
     let sub_validator_mnemonic_phrase = config.sub_validator_mnemonic_phrase.clone();
     let sub_api_url = config.sub_api_url.clone();
-    log::debug!("handle_EthRelayMessage");
+    log::debug!("handle_EthApproveRelayMessage");
 
     task_executor.spawn(lazy(move || {
         poll_fn(move || {
@@ -464,9 +466,9 @@ fn handle_eth_validators_list_message(
     new_how_many_validators_decide: U256,
 ) {
     let message_id = primitives::H256::from_slice(&message_id.to_fixed_bytes());
-    let new_validators = new_validators
+    let new_validators: Vec<AccountId32> = new_validators
         .iter()
-        .map(|a| primitives::sr25519::Public::from_slice(&a.to_fixed_bytes()))
+        .map(|a| AccountId32::from(a.to_fixed_bytes()))
         .collect::<Vec<_>>();
     let sub_validator_mnemonic_phrase = config.sub_validator_mnemonic_phrase.clone();
     let sub_api_url = config.sub_api_url.clone();
@@ -513,6 +515,34 @@ fn handle_sub_relay_message(config: &Config, task_executor: TaskExecutor, messag
     }));
 }
 
+fn handle_oracle_message(
+    config: &Config,
+    task_executor: TaskExecutor,
+    token: Vec<u8>,
+    price: Balance,
+) {
+    let sub_validator_mnemonic_phrase = config.sub_validator_mnemonic_phrase.clone();
+    let sub_api_url = config.sub_api_url.clone();
+
+    task_executor.spawn(lazy(move || {
+        poll_fn(move || {
+            blocking(|| {
+                substrate_transactions::record_price(
+                    sub_api_url.clone(),
+                    sub_validator_mnemonic_phrase.clone(),
+                    token.clone(),
+                    price,
+                );
+                log::info!(
+                    "[substrate] called record_price({:?}, {})",
+                    std::str::from_utf8(&token), price
+                );
+            })
+            .map_err(|_| panic!("the threadpool shut down"))
+        })
+    }));
+}
+
 fn handle_sub_approved_relay_message<T>(
     config: &Config,
     task_executor: TaskExecutor,

src/graph_node_event_listener.rs

@@ -956,7 +956,6 @@ fn parse_u128(maybe_u128: &str) -> u128 {
 fn parse_u256(maybe_u256: &str) -> U256 {
     let maybe_u128: u128 = maybe_u256.parse().expect("can not parse str to u128");
     let u256 = U256::from(maybe_u128);
-    log::debug!("Graph amount:{:?}, native amount: {:?}, parsed amount:{:?}", maybe_u256, maybe_u128, u256);
     u256
 }