Do Proof of History verification before appending entries to the bank

Note: replicate_stage is still using `process_entries()` because
changing it to `process_blocks()` causes the `test_replicate` test to
fail.

Cargo.toml

@@ -73,3 +73,4 @@ pnet_datalink = "0.21.0"
 tokio = "0.1"
 tokio-codec = "0.1"
 tokio-io = "0.1"
+itertools = "0.7.8"

src/bank.rs

@@ -8,6 +8,8 @@ extern crate libc;
 use chrono::prelude::*;
 use entry::Entry;
 use hash::Hash;
+use itertools::Itertools;
+use ledger::Block;
 use mint::Mint;
 use payment_plan::{Payment, PaymentPlan, Witness};
 use signature::{KeyPair, PublicKey, Signature};
@@ -28,6 +30,8 @@ use transaction::{Instruction, Plan, Transaction};
 /// not be processed by the network.
 pub const MAX_ENTRY_IDS: usize = 1024 * 16;
 
+pub const VERIFY_BLOCK_SIZE: usize = 16;
+
 /// Reasons a transaction might be rejected.
 #[derive(Debug, PartialEq, Eq)]
 pub enum BankError {
@@ -51,6 +55,9 @@ pub enum BankError {
     /// The transaction is invalid and has requested a debit or credit of negative
     /// tokens.
     NegativeTokens,
+
+    /// Proof of History verification failed.
+    LedgerVerificationFailed,
 }
 
 pub type Result<T> = result::Result<T, BankError>;
@@ -89,10 +96,9 @@ pub struct Bank {
     transaction_count: AtomicUsize,
 }
 
-impl Bank {
-    /// Create an Bank using a deposit.
-    pub fn new_from_deposit(deposit: &Payment) -> Self {
-        let bank = Bank {
+impl Default for Bank {
+    fn default() -> Self {
+        Bank {
             balances: RwLock::new(HashMap::new()),
             pending: RwLock::new(HashMap::new()),
             last_ids: RwLock::new(VecDeque::new()),
@@ -100,7 +106,14 @@ impl Bank {
             time_sources: RwLock::new(HashSet::new()),
             last_time: RwLock::new(Utc.timestamp(0, 0)),
             transaction_count: AtomicUsize::new(0),
-        };
+        }
+    }
+}
+
+impl Bank {
+    /// Create an Bank using a deposit.
+    pub fn new_from_deposit(deposit: &Payment) -> Self {
+        let bank = Self::default();
         bank.apply_payment(deposit, &mut bank.balances.write().unwrap());
         bank
     }
@@ -322,6 +335,57 @@ impl Bank {
         Ok(entry_count)
     }
 
+    /// Append entry blocks to the ledger, verifying them along the way.
+    pub fn process_blocks<I>(&self, entries: I) -> Result<u64>
+    where
+        I: IntoIterator<Item = Entry>,
+    {
+        // Ledger verification needs to be parallelized, but we can't pull the whole
+        // thing into memory. We therefore chunk it.
+        let mut entry_count = 0;
+        for block in &entries.into_iter().chunks(VERIFY_BLOCK_SIZE) {
+            let block: Vec<_> = block.collect();
+            if !block.verify(&self.last_id()) {
+                return Err(BankError::LedgerVerificationFailed);
+            }
+            entry_count += self.process_entries(block)?;
+        }
+        Ok(entry_count)
+    }
+
+    /// Process a full ledger.
+    pub fn process_ledger<I>(&self, entries: I) -> Result<u64>
+    where
+        I: IntoIterator<Item = Entry>,
+    {
+        let mut entries = entries.into_iter();
+
+        // The first item in the ledger is required to be an entry with zero num_hashes,
+        // which implies its id can be used as the ledger's seed.
+        let entry0 = entries.next().expect("invalid ledger: empty");
+
+        // The second item in the ledger is a special transaction where the to and from
+        // fields are the same. That entry should be treated as a deposit, not a
+        // transfer to oneself.
+        let entry1 = entries
+            .next()
+            .expect("invalid ledger: need at least 2 entries");
+        let tx = &entry1.transactions[0];
+        let deposit = if let Instruction::NewContract(contract) = &tx.instruction {
+            contract.plan.final_payment()
+        } else {
+            None
+        }.expect("invalid ledger, needs to start with a contract");
+
+        self.apply_payment(&deposit, &mut self.balances.write().unwrap());
+        self.register_entry_id(&entry0.id);
+        self.register_entry_id(&entry1.id);
+
+        let mut entry_count = 2;
+        entry_count += self.process_blocks(entries)?;
+        Ok(entry_count)
+    }
+
     /// Process a Witness Signature. Any payment plans waiting on this signature
     /// will progress one step.
     fn apply_signature(&self, from: PublicKey, tx_sig: Signature) -> Result<()> {

src/bin/fullnode.rs

@@ -10,9 +10,7 @@ use getopts::Options;
 use solana::bank::Bank;
 use solana::crdt::ReplicatedData;
 use solana::entry::Entry;
-use solana::payment_plan::PaymentPlan;
 use solana::server::Server;
-use solana::transaction::Instruction;
 use std::env;
 use std::fs::File;
 use std::io::{stdin, stdout, BufRead, Write};
@@ -69,7 +67,7 @@ fn main() {
 
     eprintln!("Initializing...");
     let stdin = stdin();
-    let mut entries = stdin.lock().lines().map(|line| {
+    let entries = stdin.lock().lines().map(|line| {
         let entry: Entry = serde_json::from_str(&line.unwrap()).unwrap_or_else(|e| {
             eprintln!("failed to parse json: {}", e);
             exit(1);
@@ -78,34 +76,14 @@ fn main() {
     });
     eprintln!("done parsing...");
 
-    // The first item in the ledger is required to be an entry with zero num_hashes,
-    // which implies its id can be used as the ledger's seed.
-    let entry0 = entries.next().expect("invalid ledger: empty");
-
-    // The second item in the ledger is a special transaction where the to and from
-    // fields are the same. That entry should be treated as a deposit, not a
-    // transfer to oneself.
-    let entry1 = entries
-        .next()
-        .expect("invalid ledger: need at least 2 entries");
-    let tx = &entry1.transactions[0];
-    let deposit = if let Instruction::NewContract(contract) = &tx.instruction {
-        contract.plan.final_payment()
-    } else {
-        None
-    }.expect("invalid ledger, needs to start with a contract");
-
     eprintln!("creating bank...");
-
-    let bank = Bank::new_from_deposit(&deposit);
-    bank.register_entry_id(&entry0.id);
-    bank.register_entry_id(&entry1.id);
+    let bank = Bank::default();
 
     // entry_height is the network-wide agreed height of the ledger.
     //  initialize it from the input ledger
-    eprintln!("processing entries...");
-    let entry_height = bank.process_entries(entries).expect("process_entries");
-    eprintln!("processed {} entries...", entry_height);
+    eprintln!("processing ledger...");
+    let entry_height = bank.process_ledger(entries).expect("process_ledger");
+    eprintln!("processed {} ledger...", entry_height);
 
     eprintln!("creating networking stack...");
 

src/lib.rs

@@ -52,6 +52,7 @@ extern crate bincode;
 extern crate byteorder;
 extern crate chrono;
 extern crate generic_array;
+extern crate itertools;
 extern crate libc;
 #[macro_use]
 extern crate log;