generate-epoch-blocks.rs

extern crate blake2;
extern crate byteorder;
extern crate ed25519_dalek;
extern crate hex;
extern crate lmdb_zero as lmdb;
extern crate nanocurrency_types;
extern crate reqwest;
extern crate serde_json;

use blake2::Blake2b;
use byteorder::{ByteOrder, BigEndian};
use ed25519_dalek::{Keypair, PublicKey, SecretKey};
use lmdb::LmdbResultExt;
use nanocurrency_types::{Account, Block, BlockHash, BlockHeader, BlockInner};
use std::collections::{HashSet, HashMap};
use std::thread::sleep;
use std::time::Duration;
use std::{env, io};

fn main() {
    let stdin = io::stdin();
    let mut skey_str = String::new();
    stdin
        .read_line(&mut skey_str)
        .expect("Failed to read from stdin");
    if skey_str.ends_with('\n') {
        skey_str.pop();
    }
    if skey_str.ends_with('\r') {
        skey_str.pop();
    }
    let secret_key = SecretKey::from_bytes(&hex::decode(skey_str).expect("stdin is not valid hex (should be a private key)"))
        .expect("Failed to decode private key from stdin (wrong length?)");
    let public_key = PublicKey::from_secret::<Blake2b>(&secret_key);
    let keypair = Keypair {
        secret: secret_key,
        public: public_key,
    };
    let mut args = env::args();
    args.next();
    let env = unsafe {
        let mut builder = lmdb::EnvBuilder::new().unwrap();
        builder.set_maxdbs(64).unwrap();
        builder
            .open(
                &args.next().expect("Expected path as arg"),
                lmdb::open::NOSUBDIR | lmdb::open::NOTLS,
                0o600,
            )
            .unwrap()
    };
    let rpc = args.next().expect("Expected RPC as second argument");
    let work_rpc = args.next().unwrap_or_else(|| rpc.clone());
    let open_db =
        lmdb::Database::open(&env, Some("open"), &lmdb::DatabaseOptions::defaults()).unwrap();
    let change_db =
        lmdb::Database::open(&env, Some("change"), &lmdb::DatabaseOptions::defaults()).unwrap();
    let state_db =
        lmdb::Database::open(&env, Some("state"), &lmdb::DatabaseOptions::defaults()).unwrap();
    let accounts_v0_db =
        lmdb::Database::open(&env, Some("accounts"), &lmdb::DatabaseOptions::defaults()).unwrap();
    let accounts_v1_db =
        lmdb::Database::open(&env, Some("accounts_v1"), &lmdb::DatabaseOptions::defaults()).unwrap();
    let pending_v0_db =
        lmdb::Database::open(&env, Some("pending"), &lmdb::DatabaseOptions::defaults()).unwrap();
    let mut epoch_link = [0u8; 32];
    for (i, o) in b"epoch v1 block".iter().zip(epoch_link.iter_mut()) {
        *o = *i;
    }
    let req_client = reqwest::Client::builder()
        .timeout(Duration::from_secs(30)) // for work generation
        .build()
        .expect("Failed to build reqwest client");
    let mut blocks: Vec<BlockInner> = Vec::new();
    let mut has_outdated = true;
    while has_outdated {
        {
            has_outdated = false;
            let txn = lmdb::ReadTransaction::new(&env).unwrap();
            let access = txn.access();
            let mut accounts_it = txn.cursor(&accounts_v0_db).unwrap();
            let mut current_kv = accounts_it.first::<[u8], [u8]>(&access).to_opt().unwrap();
            while let Some((account_slice, account_info)) = current_kv {
                let mut account = [0u8; 32];
                account.clone_from_slice(account_slice);
                let mut head_block = [0u8; 32];
                head_block.clone_from_slice(&account_info[..32]);
                let rep_block = &account_info[32..64];
                let mut representative = None;
                if let Some(open_block) =
                    access.get::<_, [u8]>(&open_db, rep_block).to_opt().unwrap()
                {
                    let mut rep_bytes = [0u8; 32];
                    rep_bytes.copy_from_slice(&open_block[32..64]);
                    representative = Some(rep_bytes);
                } else if let Some(change_block) = access
                    .get::<_, [u8]>(&change_db, rep_block)
                    .to_opt()
                    .unwrap()
                {
                    let mut rep_bytes = [0u8; 32];
                    rep_bytes.copy_from_slice(&change_block[32..64]);
                    representative = Some(rep_bytes);
                } else if let Some(state_block) = access
                    .get::<_, [u8]>(&state_db, rep_block)
                    .to_opt()
                    .unwrap()
                {
                    let mut rep_bytes = [0u8; 32];
                    rep_bytes.copy_from_slice(&state_block[64..96]);
                    representative = Some(rep_bytes);
                }
                let representative = representative.expect("Representative block doesn't exist");
                let balance = &account_info[96..112];
                blocks.push(BlockInner::State {
                    account: Account(account),
                    previous: BlockHash(head_block),
                    representative: Account(representative),
                    balance: BigEndian::read_u128(balance),
                    link: epoch_link,
                });
                has_outdated = true;
                current_kv = accounts_it.next::<[u8], [u8]>(&access).to_opt().unwrap();
            }
            let mut pending_it = txn.cursor(&pending_v0_db).unwrap();
            current_kv = pending_it.first::<[u8], [u8]>(&access).to_opt().unwrap();
            let mut seen_destinations = HashSet::new();
            while let Some((pending_key, _pending_info)) = current_kv {
                let destination = &pending_key[..32];
                if destination != &[0u8; 32] && seen_destinations.insert(destination) {
                    let v0_acct_exists = access
                        .get::<_, [u8]>(&accounts_v0_db, destination)
                        .to_opt()
                        .unwrap()
                        .is_some();
                    let v1_acct_exists = access
                        .get::<_, [u8]>(&accounts_v1_db, destination)
                        .to_opt()
                        .unwrap()
                        .is_some();
                    if !v0_acct_exists && !v1_acct_exists {
                        let mut destination_bytes = [0u8; 32];
                        destination_bytes.copy_from_slice(destination);
                        blocks.push(BlockInner::State {
                            account: Account(destination_bytes),
                            previous: BlockHash([0u8; 32]),
                            representative: Account([0u8; 32]),
                            balance: 0,
                            link: epoch_link,
                        });
                        has_outdated = true;
                    }
                }
                current_kv = pending_it.next::<[u8], [u8]>(&access).to_opt().unwrap();
            }
        }
        let num_blocks = blocks.len();
        for block_inner in blocks.drain(..) {
            let hash = block_inner.get_hash().0;
            let signature = keypair.sign::<Blake2b>(&hash);
            let work = {
                let root = block_inner.root_bytes();
                let root_string = hex::encode_upper(root);
                let mut args = HashMap::new();
                args.insert("action", "work_generate");
                args.insert("hash", &root_string);
                let mut res = req_client
                    .post(&work_rpc)
                    .json(&args)
                    .send()
                    .expect("Failed to send work_generate request to RPC")
                    .json::<HashMap<String, String>>()
                    .expect("Failed to parse RPC work_generate response");
                if let Some(error) = res.remove("error") {
                    panic!("RPC work_generate returned error: {}", error);
                }
                u64::from_str_radix(
                    &res.remove("work")
                        .expect("RPC work_generate response has no work field"),
                    16,
                ).expect("Failed to decode RPC work_generate response work field")
            };
            let block_header = BlockHeader { work, signature };
            let block = Block {
                header: block_header,
                inner: block_inner,
            };
            let block_string = serde_json::to_string(&block).expect("Failed to serialize block");
            let mut args = HashMap::new();
            args.insert("action", "process");
            args.insert("block", &block_string);
            let mut res = req_client
                .post(&rpc)
                .json(&args)
                .send()
                .expect("Failed to send publish request to RPC")
                .json::<HashMap<String, String>>()
                .expect("Failed to parse RPC work_generate response");
            if let Some(err) = res.remove("error") {
                eprintln!(
                    "Error processing epoch block {}:\n{}",
                    err,
                    serde_json::to_string_pretty(&block).expect("Failed to serialize block")
                );
            }
        }
        sleep(Duration::from_secs(
            5 + 2 * (num_blocks as f32).sqrt() as u64,
        ));
    }
}