Implements darwinia API (#3)

* feat(API): init darwinia API

* feat(darwinia): stable last_confirmed API

* feat(api): equip submit_proposal api with borrow error

* fix(pool): add mutex lock to transaction pool

* fix(pool): drop lock in relay service to avoid dead lock

* feat(darwinia): add relay_proposals api

* feat(darwinia): check the current proposals while submitting proposal

* feat(darwinia): add should_redeem api

* chore(git): update the Cargo.lock

Cargo.toml

@@ -41,9 +41,9 @@ features = ["full"]
 
 [dependencies.primitives]
 package = "darwinia-bridge-primitives"
-version = "0.0.8"
+# version = "0.0.8"
 git = "https://github.com/darwinia-network/bridge-primitives.git"
-branch = "types"
+branch = "checker"
 # path = "../bridge-primitives"
 features = ["runtime"]
 

src/api/darwinia.rs

@@ -1 +1,148 @@
+//! Darwinia API
+use crate::{
+    pool::EthereumTransaction,
+    result::{Error, Result},
+    Config,
+};
+use primitives::{
+    chain::eth::{HeaderStuff, PendingHeader},
+    frame::ethereum::{
+        backing::VerifiedProofStoreExt,
+        game::{PendingHeadersStoreExt, RelayProposalT, RelayProposalsStoreExt},
+        relay::{ConfirmedBlockNumbersStoreExt, SubmitProposalCallExt},
+    },
+    runtime::DarwiniaRuntime,
+};
+use sp_keyring::sr25519::sr25519::Pair;
+use substrate_subxt::{sp_core::Pair as PairTrait, Client, ClientBuilder, PairSigner};
+use web3::types::H256;
 
+/// Dawrinia API
+pub struct Darwinia {
+    client: Client<DarwiniaRuntime>,
+    /// Keyring signer
+    pub signer: PairSigner<DarwiniaRuntime, Pair>,
+}
+
+impl Darwinia {
+    /// New darwinia API
+    pub async fn new(config: &Config) -> Result<Darwinia> {
+        let pair = Pair::from_string(&config.seed, None).unwrap();
+        let signer = PairSigner::<DarwiniaRuntime, Pair>::new(pair);
+        let client = ClientBuilder::<DarwiniaRuntime>::new()
+            .set_url(&config.node)
+            .build()
+            .await?;
+
+        Ok(Darwinia { client, signer })
+    }
+
+    /// Get relay proposals
+    pub async fn relay_proposals(&self) -> Result<Vec<RelayProposalT>> {
+        Ok(self.client.relay_proposals(None).await?)
+    }
+
+    /// Get current proposals
+    pub async fn current_proposals(&self) -> Result<Vec<u64>> {
+        let proposals = self.relay_proposals().await?;
+        let mut blocks = vec![];
+        for p in proposals {
+            blocks.append(
+                &mut p
+                    .bonded_proposal
+                    .iter()
+                    .map(|bp| bp.1.header.number)
+                    .collect(),
+            )
+        }
+
+        Ok(blocks)
+    }
+
+    /// Get confirmed block numbers
+    pub async fn confirmed_block_numbers(&self) -> Result<Vec<u64>> {
+        Ok(self.client.confirmed_block_numbers(None).await?)
+    }
+
+    /// Get the last confirmed block
+    pub async fn last_confirmed(&self) -> Result<u64> {
+        Ok(
+            if let Some(confirmed) = self.confirmed_block_numbers().await?.iter().max() {
+                *confirmed
+            } else {
+                0
+            },
+        )
+    }
+
+    /// Get pending headers
+    pub async fn pending_headers(&self) -> Result<Vec<PendingHeader>> {
+        Ok(self.client.pending_headers(None).await?)
+    }
+
+    /// Submit Proposal
+    pub async fn submit_proposal(&self, proposal: Vec<HeaderStuff>) -> Result<H256> {
+        Ok(self.client.submit_proposal(&self.signer, proposal).await?)
+    }
+
+    /// Check if should redeem
+    pub async fn should_redeem(&self, tx: EthereumTransaction) -> Result<()> {
+        if let Some(res) = self
+            .client
+            .verified_proof(tx.hash(), tx.index, None)
+            .await?
+        {
+            if res {
+                Err(Error::Bridger(format!(
+                    "The tx {:?} has been redeemed",
+                    tx.hash,
+                )))
+            } else {
+                Ok(())
+            }
+        } else {
+            Ok(())
+        }
+    }
+
+    /// Check if should relay
+    pub async fn should_relay(&self, target: u64) -> Result<u64> {
+        let last_confirmed = self.last_confirmed().await?;
+        if target <= last_confirmed {
+            return Err(Error::Bridger(format!(
+                "The target block {} is not greater than the last confirmed {}",
+                target, last_confirmed,
+            )));
+        }
+
+        // Check if confirmed
+        let confirmed_blocks = self.confirmed_block_numbers().await?;
+        if confirmed_blocks.contains(&target) {
+            return Err(Error::Bridger(format!(
+                "The target block {} has already been submitted",
+                target,
+            )));
+        }
+
+        // Check if the target block is pending
+        let pending_headers = self.pending_headers().await?;
+        for p in pending_headers {
+            if p.1 == target {
+                return Err(Error::Bridger(format!(
+                    "The target block {} is pending",
+                    target,
+                )));
+            }
+        }
+
+        // Check if the target block is in relayer game
+        let proposals = self.current_proposals().await?;
+        if proposals.contains(&target) {
+            return Err(Error::Bridger(format!(
+                "The target block {} has been in relayer game",
+                target,
+            )));
+        }
+        Ok(last_confirmed)
+    }
+}

src/api/mod.rs

@@ -1,4 +1,5 @@
 //! Briger APIs
+mod darwinia;
 mod shadow;
 
-pub use shadow::Shadow;
+pub use self::{darwinia::Darwinia, shadow::Shadow};

src/api/shadow.rs

@@ -1,5 +1,5 @@
 //! Darwinia shadow API
-use crate::result::Result;
+use crate::{result::Result, Config};
 use primitives::chain::eth::{
     EthereumReceiptProofThing, EthereumReceiptProofThingJson, HeaderStuff, HeaderStuffJson,
     HeaderThing, HeaderThingWithConfirmationJson,
@@ -21,10 +21,18 @@ pub struct Shadow {
     /// Shadow API
     pub api: String,
     /// HTTP Client
-    pub client: Client,
+    pub http: Client,
 }
 
 impl Shadow {
+    /// Init Shadow API from config
+    pub fn new(config: &Config) -> Shadow {
+        Shadow {
+            api: config.shadow.clone(),
+            http: Client::new(),
+        }
+    }
+
     /// Get HeaderThing
     ///
     /// ```
@@ -45,7 +53,7 @@ impl Shadow {
     /// ```
     pub async fn header_thing(&self, number: usize) -> Result<HeaderThing> {
         let json: HeaderThingWithConfirmationJson = self
-            .client
+            .http
             .get(&format!("{}/eth/header/{}", &self.api, number))
             .send()
             .await?
@@ -75,7 +83,7 @@ impl Shadow {
     /// ```
     pub async fn receipt(&self, tx: &str) -> Result<EthereumReceiptProofThing> {
         let json: EthereumReceiptProofThingJson = self
-            .client
+            .http
             .get(&format!("{}/eth/receipt/{}", &self.api, tx))
             .send()
             .await?
@@ -104,14 +112,18 @@ impl Shadow {
     /// }
     /// ```
     pub async fn proposal(&self, member: u64, target: u64, last_leaf: u64) -> Result<HeaderStuff> {
+        info!(
+            "Requesting proposal - member: {}, target: {}, last_leaf: {}",
+            member, target, last_leaf
+        );
         let map: Value = serde_json::to_value(Proposal {
             member,
             target,
             last_leaf,
         })?;
 
         let json: HeaderStuffJson = self
-            .client
+            .http
             .post(&format!("{}/eth/proposal", self.api))
             .json(&map)
             .send()

src/cmd/run.rs

@@ -12,7 +12,7 @@ pub async fn exec(path: Option<PathBuf>, verbose: bool) -> Result<()> {
     }
     env_logger::init();
 
-    let mut listener = Listener::from_config(Config::new(path)?)?;
+    let mut listener = Listener::from_config(Config::new(path)?).await?;
     listener.start().await?;
     Ok(())
 }

src/config.rs

@@ -43,7 +43,10 @@ pub struct EthereumConfig {
 /// Service step
 #[derive(Debug, Serialize, Deserialize)]
 pub struct Step {
-    ethereum: usize,
+    /// Ethereum step
+    pub ethereum: u64,
+    /// Relay Step
+    pub relay: u64,
 }
 
 /// Bridger Config
@@ -101,7 +104,10 @@ impl Default for Config {
                     },
                 },
             },
-            step: Step { ethereum: 30 },
+            step: Step {
+                ethereum: 30,
+                relay: 60,
+            },
         }
     }
 }

src/lib.rs

@@ -6,7 +6,6 @@ extern crate log;
 
 mod config;
 mod listener;
-mod runtime;
 
 pub mod api;
 pub mod cmd;

src/listener.rs

@@ -1,11 +1,12 @@
 //! Bridger Listener
 use crate::{
+    api::{Darwinia, Shadow},
     pool::Pool,
     result::{Error, Result},
-    service::{EthereumService, Service},
+    service::{EthereumService, RelayService, Service},
     Config,
 };
-use std::{cell::RefCell, sync::Arc};
+use std::sync::{Arc, Mutex};
 use web3::transports::http::Http;
 
 /// Bridger listener
@@ -34,10 +35,10 @@ impl Listener {
 
     /// Start services
     pub async fn start(&mut self) -> Result<()> {
-        let pool = Arc::new(RefCell::new(Pool::default()));
+        let pool = Arc::new(Mutex::new(Pool::default()));
         let result = futures::future::join_all(self.0.iter_mut().map(|s| {
             info!("Start service {}", s.name());
-            s.run(pool.clone())
+            s.run(Arc::clone(&pool))
         }))
         .await;
         for r in result {
@@ -47,16 +48,26 @@ impl Listener {
     }
 
     /// Generate listener from `Config`
-    pub fn from_config(config: Config) -> Result<Self> {
+    pub async fn from_config(config: Config) -> Result<Self> {
         let mut l = Self::default();
         if config.eth.rpc.starts_with("ws") {
             return Err(Error::Bridger(
                 "Bridger currently doesn't support ethereum websocket transport".to_string(),
             ));
         }
 
+        // APIs
+        let shadow = Arc::new(Shadow::new(&config));
+        let darwinia = Arc::new(Darwinia::new(&config).await?);
+
+        // 1. Transaction Listener
+        // 2. Relay Listener
         let http = <EthereumService<Http>>::new_http(&config)?;
+        let relay = RelayService::new(&config, shadow, darwinia);
+
+        // Register
         l.register(http)?;
+        l.register(relay)?;
         Ok(l)
     }
 }

src/pool.rs

@@ -1,17 +1,53 @@
 //! Transaction pool
+use std::cmp::{Ord, Ordering, PartialOrd};
 use web3::types::H256;
 
-/// Reedeemable Ethereum transactions
-pub enum EthereumTransaction {
+/// Ethereum transaction event with hash
+#[derive(PartialEq, Eq, Debug)]
+pub enum EthereumTransactionHash {
     /// Deposit event
     Deposit(H256),
     /// Token event
     Token(H256),
 }
 
+/// Reedeemable Ethereum transaction
+#[derive(PartialEq, Eq)]
+pub struct EthereumTransaction {
+    /// Transaction event with hash
+    pub hash: EthereumTransactionHash,
+    /// Transaction block
+    pub block: u64,
+    /// Transaction index
+    pub index: u64,
+}
+
+impl EthereumTransaction {
+    /// Get the hash
+    pub fn hash(&self) -> [u8; 32] {
+        match self.hash {
+            EthereumTransactionHash::Token(h) => h,
+            EthereumTransactionHash::Deposit(h) => h,
+        }
+        .to_fixed_bytes()
+    }
+}
+
+impl PartialOrd for EthereumTransaction {
+    fn partial_cmp(&self, o: &Self) -> Option<Ordering> {
+        self.block.partial_cmp(&o.block)
+    }
+}
+
+impl Ord for EthereumTransaction {
+    fn cmp(&self, o: &Self) -> Ordering {
+        self.block.cmp(&o.block)
+    }
+}
+
 /// Transaction pool
 #[derive(Default)]
 pub struct Pool {
     /// Ethereum transactions
-    pub eth: Vec<EthereumTransaction>,
+    pub ethereum: Vec<EthereumTransaction>,
 }