faucetd: Finalize.

bin/darkfid2/src/main.rs

@@ -360,7 +360,7 @@ async fn realmain(args: Args, ex: Arc<Executor<'_>>) -> Result<()> {
     }
 
     // Consensus protocol
-    if args.consensus {
+    if args.consensus && *darkfid.synced.lock().await {
         info!("Starting consensus P2P network");
         consensus_p2p.clone().unwrap().start(ex.clone()).await?;
         let _ex = ex.clone();
@@ -374,6 +374,8 @@ async fn realmain(args: Args, ex: Arc<Executor<'_>>) -> Result<()> {
 
         info!("Starting consensus protocol task");
         ex.spawn(proposal_task(consensus_p2p.unwrap(), state)).detach();
+    } else {
+        info!("Not starting consensus P2P network");
     }
 
     // Wait for SIGINT

bin/faucetd/src/main.rs

@@ -6,7 +6,6 @@ use async_trait::async_trait;
 use chrono::Utc;
 use easy_parallel::Parallel;
 use futures_lite::future;
-use lazy_init::Lazy;
 use log::{debug, error, info};
 use num_bigint::BigUint;
 use serde_derive::Deserialize;
@@ -17,18 +16,17 @@ use structopt_toml::StructOptToml;
 use url::Url;
 
 use darkfi::{
-    async_daemonize,
-    blockchain::{NullifierStore, RootStore},
-    cli_desc,
+    async_daemonize, cli_desc,
     consensus::{
         proto::{ProtocolSync, ProtocolTx},
         task::block_sync_task,
-        Timestamp, Tx, ValidatorState, MAINNET_GENESIS_HASH_BYTES, TESTNET_GENESIS_HASH_BYTES,
+        Timestamp, Tx, ValidatorState, ValidatorStatePtr, MAINNET_GENESIS_HASH_BYTES,
+        TESTNET_GENESIS_HASH_BYTES,
     },
-    crypto::{keypair::PublicKey, types::DrkTokenId},
+    crypto::{address::Address, keypair::PublicKey, types::DrkTokenId},
     net,
     net::P2pPtr,
-    node::{Client, State},
+    node::Client,
     rpc::{
         jsonrpc,
         jsonrpc::{
@@ -44,7 +42,7 @@ use darkfi::{
         serial::serialize,
         sleep,
     },
-    wallet::walletdb::{init_wallet, WalletPtr},
+    wallet::walletdb::init_wallet,
     Error, Result,
 };
 
@@ -120,13 +118,13 @@ struct Args {
 }
 
 pub struct Faucetd {
+    synced: Mutex<bool>, // AtomicBool is weird in Arc
+    sync_p2p: P2pPtr,
+    client: Arc<Client>,
+    validator_state: ValidatorStatePtr,
     airdrop_timeout: i64,
     airdrop_limit: BigUint,
-    airdrop_map: Arc<Mutex<HashMap<String, i64>>>,
-    synced: Mutex<bool>, // AtomicBool is weird in Arc
-    client: Client,
-    p2p: P2pPtr,
-    state: Arc<Mutex<State>>,
+    airdrop_map: Arc<Mutex<HashMap<Address, i64>>>,
 }
 
 #[async_trait]
@@ -147,39 +145,21 @@ impl RequestHandler for Faucetd {
 
 impl Faucetd {
     pub async fn new(
-        db: &sled::Db,
-        wallet: WalletPtr,
-        p2p: P2pPtr,
+        validator_state: ValidatorStatePtr,
+        sync_p2p: P2pPtr,
         timeout: i64,
         limit: BigUint,
     ) -> Result<Self> {
-        // Initialize client
-        let client = Client::new(wallet).await?;
-        let tree = client.get_tree().await?;
-        let merkle_roots = RootStore::new(db)?;
-        let nullifiers = NullifierStore::new(db)?;
-
-        let kp = client.main_keypair.lock().await.public;
-
-        // Initialize state
-        let state = Arc::new(Mutex::new(State {
-            tree,
-            merkle_roots,
-            nullifiers,
-            cashier_pubkeys: vec![],
-            faucet_pubkeys: vec![kp],
-            mint_vk: Lazy::new(),
-            burn_vk: Lazy::new(),
-        }));
+        let client = validator_state.read().await.client.clone();
 
         Ok(Self {
+            synced: Mutex::new(false),
+            sync_p2p,
+            client,
+            validator_state,
             airdrop_timeout: timeout,
             airdrop_limit: limit,
             airdrop_map: Arc::new(Mutex::new(HashMap::new())),
-            synced: Mutex::new(false),
-            client,
-            p2p,
-            state,
         })
     }
 
@@ -193,39 +173,68 @@ impl Faucetd {
             return jsonrpc::error(InvalidParams, None, id).into()
         }
 
-        let pubkey = match PublicKey::from_str(params[0].as_str().unwrap()) {
+        if *self.synced.lock().await == false {
+            error!("airdrop(): Blockchain is not yet synced");
+            return jsonrpc::error(InternalError, None, id).into()
+        }
+
+        let address = match Address::from_str(params[0].as_str().unwrap()) {
+            Ok(v) => v,
+            Err(_) => {
+                error!("airdrop(): Failed parsing address from string");
+                return server_error(RpcError::ParseError, id)
+            }
+        };
+
+        let pubkey = match PublicKey::try_from(address) {
             Ok(v) => v,
-            Err(_) => return server_error(RpcError::ParseError, id),
+            Err(_) => {
+                error!("airdrop(): Failed parsing PublicKey from Address");
+                return server_error(RpcError::ParseError, id)
+            }
         };
 
         let amount = match decode_base10(params[1].as_str().unwrap(), 8, true) {
             Ok(v) => v,
-            Err(_) => return server_error(RpcError::ParseError, id),
+            Err(_) => {
+                error!("airdrop(): Failed parsing amount from string");
+                return server_error(RpcError::ParseError, id)
+            }
         };
 
         if amount > self.airdrop_limit {
             return server_error(RpcError::AmountExceedsLimit, id)
         }
 
+        // Check if there as a previous airdrop and the timeout has passed.
         let now = Utc::now().timestamp();
         let map = self.airdrop_map.lock().await;
-        if let Some(last_airdrop) = map.get(params[1].as_str().unwrap()) {
+        if let Some(last_airdrop) = map.get(&address) {
             if now - last_airdrop <= self.airdrop_timeout {
                 return server_error(RpcError::TimeLimitReached, id)
             }
         };
         drop(map);
 
-        if *self.synced.lock().await == false {
-            error!("airdrop(): Blockchain is not yet synced");
-            return jsonrpc::error(InternalError, None, id).into()
-        }
-
         // TODO: Token ID decision
-        // TODO: Rename this function to tx build
+        let token_id = DrkTokenId::from(1);
+        let amnt: u64 = match amount.try_into() {
+            Ok(v) => v,
+            Err(e) => {
+                error!("airdrop(): Failed converting biguint to u64: {}", e);
+                return jsonrpc::error(InternalError, None, id).into()
+            }
+        };
+
         let tx = match self
             .client
-            .send(pubkey, amount.try_into().unwrap(), DrkTokenId::from(1), true, self.state.clone())
+            .build_transaction(
+                pubkey,
+                amnt,
+                token_id,
+                true,
+                self.validator_state.read().await.state_machine.clone(),
+            )
             .await
         {
             Ok(v) => v,
@@ -235,10 +244,8 @@ impl Faucetd {
             }
         };
 
-        let tx_hash = blake3::hash(&serialize(&tx)).to_hex().as_str().to_string();
-
         // Broadcast transaction to the network.
-        match self.p2p.broadcast(Tx(tx)).await {
+        match self.sync_p2p.broadcast(Tx(tx.clone())).await {
             Ok(()) => {}
             Err(e) => {
                 error!("airdrop(): Failed broadcasting transaction: {}", e);
@@ -246,15 +253,17 @@ impl Faucetd {
             }
         }
 
+        // Add/Update this airdrop into the hashmap
         let mut map = self.airdrop_map.lock().await;
-        map.insert(params[1].as_str().unwrap().to_string(), now);
+        map.insert(address, now);
         drop(map);
 
+        let tx_hash = blake3::hash(&serialize(&tx)).to_hex().as_str().to_string();
         jsonrpc::response(json!(tx_hash), id).into()
     }
 }
 
-async fn prune_airdrop_map(map: Arc<Mutex<HashMap<String, i64>>>, timeout: i64) {
+async fn prune_airdrop_map(map: Arc<Mutex<HashMap<Address, i64>>>, timeout: i64) {
     loop {
         sleep(timeout as u64).await;
         debug!("Pruning airdrop map");
@@ -310,11 +319,12 @@ async fn realmain(args: Args, ex: Arc<Executor<'_>>) -> Result<()> {
     };
 
     // TODO: sqldb init cleanup
-    Client::new(wallet.clone()).await?;
-    let address = wallet.get_default_address().await?;
+    // Initialize client
+    let client = Arc::new(Client::new(wallet).await?);
 
     // Initialize validator state
-    let state = ValidatorState::new(&sled_db, address, genesis_ts, genesis_data)?;
+    let state =
+        ValidatorState::new(&sled_db, genesis_ts, genesis_data, client, vec![], vec![]).await?;
 
     // P2P network. The faucet doesn't participate in consensus, so we only
     // build the sync protocol.
@@ -327,8 +337,8 @@ async fn realmain(args: Args, ex: Arc<Executor<'_>>) -> Result<()> {
         ..Default::default()
     };
 
-    let p2p = net::P2p::new(network_settings).await;
-    let registry = p2p.protocol_registry();
+    let sync_p2p = net::P2p::new(network_settings).await;
+    let registry = sync_p2p.protocol_registry();
 
     info!("Registering block sync P2P protocols...");
     let _state = state.clone();
@@ -352,7 +362,7 @@ async fn realmain(args: Args, ex: Arc<Executor<'_>>) -> Result<()> {
 
     // Initialize program state
     let faucetd =
-        Faucetd::new(&sled_db, wallet, p2p.clone(), airdrop_timeout, airdrop_limit).await?;
+        Faucetd::new(state.clone(), sync_p2p.clone(), airdrop_timeout, airdrop_limit).await?;
     let faucetd = Arc::new(faucetd);
 
     // Task to periodically clean up the hashmap of airdrops.
@@ -363,9 +373,9 @@ async fn realmain(args: Args, ex: Arc<Executor<'_>>) -> Result<()> {
     ex.spawn(listen_and_serve(args.rpc_listen, faucetd.clone())).detach();
 
     info!("Starting sync P2P network");
-    p2p.clone().start(ex.clone()).await?;
+    sync_p2p.clone().start(ex.clone()).await?;
     let _ex = ex.clone();
-    let _sync_p2p = p2p.clone();
+    let _sync_p2p = sync_p2p.clone();
     ex.spawn(async move {
         if let Err(e) = _sync_p2p.run(_ex).await {
             error!("Failed starting sync P2P network: {}", e);
@@ -373,7 +383,7 @@ async fn realmain(args: Args, ex: Arc<Executor<'_>>) -> Result<()> {
     })
     .detach();
 
-    match block_sync_task(p2p.clone(), state.clone()).await {
+    match block_sync_task(sync_p2p.clone(), state.clone()).await {
         Ok(()) => *faucetd.synced.lock().await = true,
         Err(e) => error!("Failed syncing blockchain: {}", e),
     }

src/crypto/address.rs

@@ -12,7 +12,7 @@ enum AddressType {
     Payment = 0,
 }
 
-#[derive(Copy, Clone, Debug, Eq, Ord, PartialEq, PartialOrd)]
+#[derive(Copy, Clone, Debug, Eq, Ord, PartialEq, PartialOrd, Hash)]
 pub struct Address(pub [u8; 37]);
 
 impl Address {

src/node/client.rs

@@ -149,10 +149,8 @@ impl Client {
         Ok((tx, coins))
     }
 
-    // TODO: This was changed so it does not broadcast transactions anymore.
-    // Instead, it returns the transaction itself, which is then able to be
-    // arbitrarily broadcasted. Rename the function from send() to a better name.
-    pub async fn send(
+    /// Build a transaction given the required parameters and state machine.
+    pub async fn build_transaction(
         &self,
         pubkey: PublicKey,
         amount: u64,
@@ -179,22 +177,23 @@ impl Client {
         Ok(tx)
     }
 
-    pub async fn transfer(
-        &self,
-        token_id: DrkTokenId,
-        pubkey: PublicKey,
-        amount: u64,
-        state: Arc<Mutex<State>>,
-    ) -> ClientResult<()> {
-        debug!("transfer(): Start transfer {}", amount);
-        if self.wallet.token_id_exists(token_id).await? {
-            self.send(pubkey, amount, token_id, false, state).await?;
-            debug!("transfer(): Finish transfer {}", amount);
-            return Ok(())
-        }
-
-        Err(ClientFailed::NotEnoughValue(amount))
-    }
+    // TODO
+    // pub async fn transfer(
+    // &self,
+    // token_id: DrkTokenId,
+    // pubkey: PublicKey,
+    // amount: u64,
+    // state: Arc<Mutex<State>>,
+    // ) -> ClientResult<()> {
+    // debug!("transfer(): Start transfer {}", amount);
+    // if self.wallet.token_id_exists(token_id).await? {
+    // self.send(pubkey, amount, token_id, false, state).await?;
+    // debug!("transfer(): Finish transfer {}", amount);
+    // return Ok(())
+    // }
+    //
+    //      Err(ClientFailed::NotEnoughValue(amount))
+    //}
 
     pub async fn init_db(&self) -> Result<()> {
         self.wallet.init_db().await