Implement transaction pool basics

Cargo.toml

@@ -3,6 +3,7 @@ members = [
 	"eth2/attester",
 	"eth2/block_proposer",
 	"eth2/fork_choice",
+	"eth2/operation_pool",
 	"eth2/state_processing",
 	"eth2/types",
 	"eth2/utils/bls",

eth2/operation_pool/Cargo.toml

@@ -0,0 +1,9 @@
+[package]
+name = "operation_pool"
+version = "0.1.0"
+authors = ["Michael Sproul <michael@sigmaprime.io>"]
+edition = "2018"
+
+[dependencies]
+types = { path = "../types" }
+state_processing = { path = "../../eth2/state_processing" }

eth2/operation_pool/src/lib.rs

@@ -0,0 +1,308 @@
+use std::collections::{btree_map::Entry, BTreeMap, HashSet};
+
+use state_processing::per_block_processing::{
+    verify_deposit_merkle_proof, verify_exit, verify_proposer_slashing, verify_transfer,
+    verify_transfer_partial,
+};
+use types::{
+    AttesterSlashing, BeaconState, ChainSpec, Deposit, ProposerSlashing, Transfer, VoluntaryExit,
+};
+
+#[cfg(test)]
+const VERIFY_DEPOSIT_PROOFS: bool = false;
+#[cfg(not(test))]
+const VERIFY_DEPOSIT_PROOFS: bool = true;
+
+#[derive(Default)]
+pub struct OperationPool {
+    /// Map from deposit index to deposit data.
+    // NOTE: We assume that there is only one deposit per index
+    // because the Eth1 data is updated (at most) once per epoch,
+    // and the spec doesn't seem to accomodate for re-orgs on a time-frame
+    // longer than an epoch
+    deposits: BTreeMap<u64, Deposit>,
+    /// Map from attester index to slashing.
+    attester_slashings: BTreeMap<u64, AttesterSlashing>,
+    /// Map from proposer index to slashing.
+    proposer_slashings: BTreeMap<u64, ProposerSlashing>,
+    /// Map from exiting validator to their exit data.
+    voluntary_exits: BTreeMap<u64, VoluntaryExit>,
+    /// Set of transfers.
+    transfers: HashSet<Transfer>,
+}
+
+#[derive(Debug, PartialEq, Clone)]
+pub enum DepositInsertStatus {
+    /// The deposit was not already in the pool.
+    Fresh,
+    /// The deposit already existed in the pool.
+    Duplicate,
+    /// The deposit conflicted with an existing deposit, which was replaced.
+    Replaced(Deposit),
+}
+
+impl OperationPool {
+    /// Create a new operation pool.
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Add a deposit to the pool.
+    ///
+    /// No two distinct deposits should be added with the same index.
+    pub fn insert_deposit(&mut self, deposit: Deposit) -> DepositInsertStatus {
+        use DepositInsertStatus::*;
+
+        match self.deposits.entry(deposit.index) {
+            Entry::Vacant(entry) => {
+                entry.insert(deposit);
+                Fresh
+            }
+            Entry::Occupied(mut entry) => {
+                if entry.get() == &deposit {
+                    Duplicate
+                } else {
+                    Replaced(entry.insert(deposit))
+                }
+            }
+        }
+    }
+
+    /// Get an ordered list of deposits for inclusion in a block.
+    ///
+    /// Take at most the maximum number of deposits, beginning from the current deposit index.
+    pub fn get_deposits(&self, state: &BeaconState, spec: &ChainSpec) -> Vec<Deposit> {
+        let start_idx = state.deposit_index;
+        (start_idx..start_idx + spec.max_deposits)
+            .map(|idx| self.deposits.get(&idx))
+            .take_while(|deposit| {
+                // NOTE: we don't use verify_deposit, because it requires the
+                // deposit's index to match the state's, and we would like to return
+                // a batch with increasing indices
+                deposit.map_or(false, |deposit| {
+                    !VERIFY_DEPOSIT_PROOFS || verify_deposit_merkle_proof(state, deposit, spec)
+                })
+            })
+            .flatten()
+            .cloned()
+            .collect()
+    }
+
+    /// Remove all deposits with index less than the deposit index of the latest finalised block.
+    pub fn prune_deposits(&mut self, state: &BeaconState) -> BTreeMap<u64, Deposit> {
+        let deposits_keep = self.deposits.split_off(&state.deposit_index);
+        std::mem::replace(&mut self.deposits, deposits_keep)
+    }
+
+    /// Insert a proposer slashing into the pool.
+    pub fn insert_proposer_slashing(
+        &mut self,
+        slashing: ProposerSlashing,
+        state: &BeaconState,
+        spec: &ChainSpec,
+    ) -> Result<(), ()> {
+        // TODO: should maybe insert anyway if the proposer is unknown in the validator index,
+        // because they could *become* known later
+        // FIXME: error handling
+        verify_proposer_slashing(&slashing, state, spec).map_err(|_| ())?;
+        self.proposer_slashings
+            .insert(slashing.proposer_index, slashing);
+        Ok(())
+    }
+
+    /// Only check whether the implicated validator has already been slashed, because
+    /// all slashings in the pool were validated upon insertion.
+    // TODO: we need a mechanism to avoid including a proposer slashing and an attester
+    // slashing for the same validator in the same block
+    pub fn get_proposer_slashings(
+        &self,
+        state: &BeaconState,
+        spec: &ChainSpec,
+    ) -> Vec<ProposerSlashing> {
+        // We sort by validator index, which is safe, because a validator can only supply
+        // so many valid slashings for lower-indexed validators (and even that is unlikely)
+        filter_limit_operations(
+            self.proposer_slashings.values(),
+            |slashing| {
+                state
+                    .validator_registry
+                    .get(slashing.proposer_index as usize)
+                    .map_or(false, |validator| !validator.slashed)
+            },
+            spec.max_proposer_slashings,
+        )
+    }
+
+    /// Prune slashings for all slashed or withdrawn validators.
+    pub fn prune_proposer_slashings(&mut self, finalized_state: &BeaconState, spec: &ChainSpec) {
+        let to_prune = self
+            .proposer_slashings
+            .keys()
+            .flat_map(|&validator_index| {
+                finalized_state
+                    .validator_registry
+                    .get(validator_index as usize)
+                    .filter(|validator| {
+                        validator.slashed
+                            || validator.is_withdrawable_at(finalized_state.current_epoch(spec))
+                    })
+                    .map(|_| validator_index)
+            })
+            .collect::<Vec<_>>();
+
+        for validator_index in to_prune {
+            self.proposer_slashings.remove(&validator_index);
+        }
+    }
+
+    // TODO: copy ProposerSlashing code for AttesterSlashing
+
+    /// Insert a voluntary exit, validating it almost-entirely (future exits are permitted).
+    pub fn insert_voluntary_exit(
+        &mut self,
+        exit: VoluntaryExit,
+        state: &BeaconState,
+        spec: &ChainSpec,
+    ) -> Result<(), ()> {
+        verify_exit(state, &exit, spec, false).map_err(|_| ())?;
+        self.voluntary_exits.insert(exit.validator_index, exit);
+        Ok(())
+    }
+
+    /// Get a list of voluntary exits for inclusion in a block.
+    // TODO: could optimise this by eliding the checks that have already been done on insert
+    pub fn get_voluntary_exits(&self, state: &BeaconState, spec: &ChainSpec) -> Vec<VoluntaryExit> {
+        filter_limit_operations(
+            self.voluntary_exits.values(),
+            |exit| verify_exit(state, exit, spec, true).is_ok(),
+            spec.max_voluntary_exits,
+        )
+    }
+
+    /// Prune if validator has already exited at the last finalized state.
+    pub fn prune_voluntary_exits(&mut self, finalized_state: &BeaconState, spec: &ChainSpec) {
+        let to_prune = self
+            .voluntary_exits
+            .keys()
+            .flat_map(|&validator_index| {
+                finalized_state
+                    .validator_registry
+                    .get(validator_index as usize)
+                    .filter(|validator| validator.is_exited_at(finalized_state.current_epoch(spec)))
+                    .map(|_| validator_index)
+            })
+            .collect::<Vec<_>>();
+
+        for validator_index in to_prune {
+            self.voluntary_exits.remove(&validator_index);
+        }
+    }
+
+    /// Insert a transfer into the pool, checking it for validity in the process.
+    pub fn insert_transfer(
+        &mut self,
+        transfer: Transfer,
+        state: &BeaconState,
+        spec: &ChainSpec,
+    ) -> Result<(), ()> {
+        // The signature of the transfer isn't hashed, but because we check
+        // it before we insert into the HashSet, we can't end up with duplicate
+        // transactions.
+        verify_transfer_partial(state, &transfer, spec, true).map_err(|_| ())?;
+        self.transfers.insert(transfer);
+        Ok(())
+    }
+
+    /// Get a list of transfers for inclusion in a block.
+    // TODO: improve the economic optimality of this function by taking the transfer
+    // fees into account, and dependencies between transfers in the same block
+    // e.g. A pays B, B pays C
+    pub fn get_transfers(&self, state: &BeaconState, spec: &ChainSpec) -> Vec<Transfer> {
+        filter_limit_operations(
+            &self.transfers,
+            |transfer| verify_transfer(state, transfer, spec).is_ok(),
+            spec.max_transfers,
+        )
+    }
+
+    /// Prune the set of transfers by removing all those whose slot has already passed.
+    pub fn prune_transfers(&mut self, finalized_state: &BeaconState) {
+        self.transfers = self
+            .transfers
+            .drain()
+            .filter(|transfer| transfer.slot > finalized_state.slot)
+            .collect();
+    }
+}
+
+/// Filter up to a maximum number of operations out of a slice.
+fn filter_limit_operations<'a, T: 'a, I, F>(operations: I, filter: F, limit: u64) -> Vec<T>
+where
+    I: IntoIterator<Item = &'a T>,
+    F: Fn(&T) -> bool,
+    T: Clone,
+{
+    operations
+        .into_iter()
+        .filter(|x| filter(*x))
+        .take(limit as usize)
+        .cloned()
+        .collect()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::DepositInsertStatus::*;
+    use super::*;
+    use types::test_utils::{SeedableRng, TestRandom, XorShiftRng};
+
+    #[test]
+    fn insert_deposit() {
+        let mut rng = XorShiftRng::from_seed([42; 16]);
+        let mut op_pool = OperationPool::new();
+        let deposit1 = Deposit::random_for_test(&mut rng);
+        let mut deposit2 = Deposit::random_for_test(&mut rng);
+        deposit2.index = deposit1.index;
+
+        assert_eq!(op_pool.insert_deposit(deposit1.clone()), Fresh);
+        assert_eq!(op_pool.insert_deposit(deposit1.clone()), Duplicate);
+        assert_eq!(op_pool.insert_deposit(deposit2), Replaced(deposit1));
+    }
+
+    #[test]
+    fn get_deposits_max() {
+        let mut rng = XorShiftRng::from_seed([42; 16]);
+        let mut op_pool = OperationPool::new();
+        let spec = ChainSpec::foundation();
+        let start = 10000;
+        let max_deposits = spec.max_deposits;
+        let extra = 5;
+        let offset = 1;
+        assert!(offset <= extra);
+
+        let proto_deposit = Deposit::random_for_test(&mut rng);
+        let deposits = (start..start + max_deposits + extra)
+            .map(|index| {
+                let mut deposit = proto_deposit.clone();
+                deposit.index = index;
+                deposit
+            })
+            .collect::<Vec<_>>();
+
+        for deposit in &deposits {
+            assert_eq!(op_pool.insert_deposit(deposit.clone()), Fresh);
+        }
+
+        let mut state = BeaconState::random_for_test(&mut rng);
+        state.deposit_index = start + offset;
+        let deposits_for_block = op_pool.get_deposits(&state, &spec);
+
+        assert_eq!(deposits_for_block.len() as u64, max_deposits);
+        assert_eq!(
+            deposits_for_block[..],
+            deposits[offset as usize..(offset + max_deposits) as usize]
+        );
+    }
+
+    // TODO: more tests
+}

eth2/state_processing/src/per_block_processing.rs

@@ -1,4 +1,3 @@
-use self::verify_proposer_slashing::verify_proposer_slashing;
 use crate::common::slash_validator;
 use errors::{BlockInvalid as Invalid, BlockProcessingError as Error, IntoWithIndex};
 use rayon::prelude::*;
@@ -8,11 +7,15 @@ use types::*;
 pub use self::verify_attester_slashing::{
     gather_attester_slashing_indices, verify_attester_slashing,
 };
+pub use self::verify_proposer_slashing::verify_proposer_slashing;
 pub use validate_attestation::{validate_attestation, validate_attestation_without_signature};
-pub use verify_deposit::{get_existing_validator_index, verify_deposit, verify_deposit_index};
+pub use verify_deposit::{
+    get_existing_validator_index, verify_deposit, verify_deposit_index,
+    verify_deposit_merkle_proof,
+};
 pub use verify_exit::verify_exit;
 pub use verify_slashable_attestation::verify_slashable_attestation;
-pub use verify_transfer::{execute_transfer, verify_transfer};
+pub use verify_transfer::{execute_transfer, verify_transfer, verify_transfer_partial};
 
 pub mod errors;
 mod validate_attestation;
@@ -426,7 +429,7 @@ pub fn process_exits(
         .par_iter()
         .enumerate()
         .try_for_each(|(i, exit)| {
-            verify_exit(&state, exit, spec).map_err(|e| e.into_with_index(i))
+            verify_exit(&state, exit, spec, true).map_err(|e| e.into_with_index(i))
         })?;
 
     // Update the state in series.

eth2/state_processing/src/per_block_processing/errors.rs

@@ -390,6 +390,11 @@ pub enum TransferInvalid {
     ///
     /// (state_slot, transfer_slot)
     StateSlotMismatch(Slot, Slot),
+    /// The `transfer.slot` is in the past relative to the state slot.
+    ///
+    ///
+    /// (state_slot, transfer_slot)
+    TransferSlotInPast(Slot, Slot),
     /// The `transfer.from` validator has been activated and is not withdrawable.
     ///
     /// (from_validator)

eth2/state_processing/src/per_block_processing/verify_deposit.rs

@@ -89,7 +89,11 @@ pub fn get_existing_validator_index(
 /// Verify that a deposit is included in the state's eth1 deposit root.
 ///
 /// Spec v0.5.0
-fn verify_deposit_merkle_proof(state: &BeaconState, deposit: &Deposit, spec: &ChainSpec) -> bool {
+pub fn verify_deposit_merkle_proof(
+    state: &BeaconState,
+    deposit: &Deposit,
+    spec: &ChainSpec,
+) -> bool {
     let leaf = hash(&get_serialized_deposit_data(deposit));
     verify_merkle_proof(
         Hash256::from_slice(&leaf),

eth2/state_processing/src/per_block_processing/verify_exit.rs

@@ -7,11 +7,17 @@ use types::*;
 ///
 /// Returns `Ok(())` if the `Exit` is valid, otherwise indicates the reason for invalidity.
 ///
+/// The `check_future_epoch` argument determines whether the exit's epoch should be checked
+/// against the state's current epoch to ensure it doesn't occur in the future.
+/// It should ordinarily be set to true, except for operations stored for
+/// some time (such as in the OperationPool).
+///
 /// Spec v0.5.0
 pub fn verify_exit(
     state: &BeaconState,
     exit: &VoluntaryExit,
     spec: &ChainSpec,
+    check_future_epoch: bool,
 ) -> Result<(), Error> {
     let validator = state
         .validator_registry
@@ -32,7 +38,7 @@ pub fn verify_exit(
 
     // Exits must specify an epoch when they become valid; they are not valid before then.
     verify!(
-        state.current_epoch(spec) >= exit.epoch,
+        !check_future_epoch || state.current_epoch(spec) >= exit.epoch,
         Invalid::FutureEpoch {
             state: state.current_epoch(spec),
             exit: exit.epoch

eth2/state_processing/src/per_block_processing/verify_transfer.rs

@@ -15,6 +15,19 @@ pub fn verify_transfer(
     state: &BeaconState,
     transfer: &Transfer,
     spec: &ChainSpec,
+) -> Result<(), Error> {
+    verify_transfer_partial(state, transfer, spec, false)
+}
+
+/// Parametric version of `verify_transfer` that allows some checks to be skipped.
+///
+/// In everywhere except the operation pool, `verify_transfer` should be preferred over this
+/// function.
+pub fn verify_transfer_partial(
+    state: &BeaconState,
+    transfer: &Transfer,
+    spec: &ChainSpec,
+    for_op_pool: bool,
 ) -> Result<(), Error> {
     let sender_balance = *state
         .validator_balances
@@ -27,17 +40,18 @@ pub fn verify_transfer(
         .ok_or_else(|| Error::Invalid(Invalid::FeeOverflow(transfer.amount, transfer.fee)))?;
 
     verify!(
-        sender_balance >= transfer.amount,
+        for_op_pool || sender_balance >= transfer.amount,
         Invalid::FromBalanceInsufficient(transfer.amount, sender_balance)
     );
 
     verify!(
-        sender_balance >= transfer.fee,
+        for_op_pool || sender_balance >= transfer.fee,
         Invalid::FromBalanceInsufficient(transfer.fee, sender_balance)
     );
 
     verify!(
-        (sender_balance == total_amount)
+            for_op_pool
+            || (sender_balance == total_amount)
             || (sender_balance >= (total_amount + spec.min_deposit_amount)),
         Invalid::InvalidResultingFromBalance(
             sender_balance - total_amount,
@@ -45,10 +59,17 @@ pub fn verify_transfer(
         )
     );
 
-    verify!(
-        state.slot == transfer.slot,
-        Invalid::StateSlotMismatch(state.slot, transfer.slot)
-    );
+    if for_op_pool {
+        verify!(
+            state.slot <= transfer.slot,
+            Invalid::TransferSlotInPast(state.slot, transfer.slot)
+        );
+    } else {
+        verify!(
+            state.slot == transfer.slot,
+            Invalid::StateSlotMismatch(state.slot, transfer.slot)
+        );
+    }
 
     let sender_validator = state
         .validator_registry
@@ -57,7 +78,8 @@ pub fn verify_transfer(
     let epoch = state.slot.epoch(spec.slots_per_epoch);
 
     verify!(
-        sender_validator.is_withdrawable_at(epoch)
+        for_op_pool
+            || sender_validator.is_withdrawable_at(epoch)
             || sender_validator.activation_epoch == spec.far_future_epoch,
         Invalid::FromValidatorIneligableForTransfer(transfer.sender)
     );

eth2/types/Cargo.toml

@@ -8,6 +8,7 @@ edition = "2018"
 bls = { path = "../utils/bls" }
 boolean-bitfield = { path = "../utils/boolean-bitfield" }
 dirs = "1.0"
+derivative = "1.0"
 ethereum-types = "0.5"
 hashing = { path = "../utils/hashing" }
 hex = "0.3"

eth2/types/src/transfer.rs

@@ -1,6 +1,7 @@
 use super::Slot;
 use crate::test_utils::TestRandom;
 use bls::{PublicKey, Signature};
+use derivative::Derivative;
 use rand::RngCore;
 use serde_derive::{Deserialize, Serialize};
 use ssz::TreeHash;
@@ -12,7 +13,6 @@ use test_random_derive::TestRandom;
 /// Spec v0.5.0
 #[derive(
     Debug,
-    PartialEq,
     Clone,
     Serialize,
     Deserialize,
@@ -21,7 +21,9 @@ use test_random_derive::TestRandom;
     TreeHash,
     TestRandom,
     SignedRoot,
+    Derivative,
 )]
+#[derivative(PartialEq, Eq, Hash)]
 pub struct Transfer {
     pub sender: u64,
     pub recipient: u64,
@@ -29,6 +31,7 @@ pub struct Transfer {
     pub fee: u64,
     pub slot: Slot,
     pub pubkey: PublicKey,
+    #[derivative(Hash = "ignore")]
     pub signature: Signature,
 }