Minimal Transaction Pool (#1067)

* verify a tx like we verify a block (experimental)

* first minimal_pool test up and running but not testing what we need to

* rework tx_pool validation to use txhashset extension

* minimal tx pool wired up but rough

* works locally (rough statew though)
delete "legacy" pool and graph code

* rework the new pool into TransactionPool and Pool impls

* rework pool to store pool entries
with associated timer and source etc.

* all_transactions

* extra_txs so we can validate stempool against existing txpool

* rework reconcile_block

* txhashset apply_raw_tx can now rewind to a checkpoint (prev raw tx)

* wip - txhashset tx tests

* more flexible rewind on MMRs

* add tests to cover apply_raw_txs on txhashset extension

* add_to_stempool and add_to_txpool

* deaggregate multi kernel tx when adding to txpoool

* handle freshness in stempool
handle propagation of stempool txs via dandelion monitor

* patience timer and fluff if we cannot propagate
to next relay

* aggregate and fluff stempool is we have no relay

* refactor coinbase maturity

* rewrote basic tx pool tests to use a real txhashset via chain adapter

* rework dandelion monitor to reflect recent discussion
works locally but needs a cleanup

* refactor dandelion_monitor - split out phases

* more pool test coverage

* remove old test code from pool (still wip)

* block_building and block_reconciliation tests

* tracked down chain test failure...

* fix test_coinbase_maturity

* dandelion_monitor now runs...

* refactor dandelion config, shared across p2p and pool components

* fix pool tests with new config

* fix p2p tests

* rework tx pool to deal with duplicate commitments (testnet2 limitation)

* cleanup and address some PR feedback

* add big comment about pre_tx...

chain/src/chain.rs

@@ -20,10 +20,10 @@ use std::fs::File;
 use std::sync::{Arc, Mutex, RwLock};
 use std::time::{Duration, Instant};
 
-use core::core::{Block, BlockHeader, Input, Output, OutputFeatures, OutputIdentifier, TxKernel};
 use core::core::hash::{Hash, Hashed};
 use core::core::pmmr::MerkleProof;
 use core::core::target::Difficulty;
+use core::core::{Block, BlockHeader, Output, OutputIdentifier, Transaction, TxKernel};
 use core::global;
 use grin_store::Error::NotFoundErr;
 use pipe;
@@ -150,9 +150,9 @@ impl Chain {
 		}
 	}
 
-	/// Initializes the blockchain and returns a new Chain instance. Does a check
-	/// on the current chain head to make sure it exists and creates one based
-	/// on the genesis block if necessary.
+	/// Initializes the blockchain and returns a new Chain instance. Does a
+	/// check on the current chain head to make sure it exists and creates one
+	/// based on the genesis block if necessary.
 	pub fn init(
 		db_root: String,
 		adapter: Arc<ChainAdapter>,
@@ -450,15 +450,57 @@ impl Chain {
 		}
 	}
 
-	/// For the given commitment find the unspent output and return the associated
-	/// Return an error if the output does not exist or has been spent.
-	/// This querying is done in a way that is consistent with the current chain state,
-	/// specifically the current winning (valid, most work) fork.
+	/// For the given commitment find the unspent output and return the
+	/// associated Return an error if the output does not exist or has been
+	/// spent. This querying is done in a way that is consistent with the
+	/// current chain state, specifically the current winning (valid, most
+	/// work) fork.
 	pub fn is_unspent(&self, output_ref: &OutputIdentifier) -> Result<Hash, Error> {
 		let mut txhashset = self.txhashset.write().unwrap();
 		txhashset.is_unspent(output_ref)
 	}
 
+	/// Validate a vector of "raw" transactions against the current chain state.
+	pub fn validate_raw_txs(
+		&self,
+		txs: Vec<Transaction>,
+		pre_tx: Option<Transaction>,
+	) -> Result<Vec<Transaction>, Error> {
+		let bh = self.head_header()?;
+		let mut txhashset = self.txhashset.write().unwrap();
+		txhashset::extending_readonly(&mut txhashset, |extension| {
+			let valid_txs = extension.validate_raw_txs(txs, pre_tx, bh.height)?;
+			Ok(valid_txs)
+		})
+	}
+
+	fn next_block_height(&self) -> Result<u64, Error> {
+		let bh = self.head_header()?;
+		Ok(bh.height + 1)
+	}
+
+	/// Verify we are not attempting to spend a coinbase output
+	/// that has not yet sufficiently matured.
+	pub fn verify_coinbase_maturity(&self, tx: &Transaction) -> Result<(), Error> {
+		let height = self.next_block_height()?;
+		let mut txhashset = self.txhashset.write().unwrap();
+		txhashset::extending_readonly(&mut txhashset, |extension| {
+			extension.verify_coinbase_maturity(&tx.inputs, height)?;
+			Ok(())
+		})
+	}
+
+	/// Verify that the tx has a lock_height that is less than or equal to
+	/// the height of the next block.
+	pub fn verify_tx_lock_height(&self, tx: &Transaction) -> Result<(), Error> {
+		let height = self.next_block_height()?;
+		if tx.lock_height() <= height {
+			Ok(())
+		} else {
+			Err(Error::TxLockHeight)
+		}
+	}
+
 	/// Validate the current chain state.
 	pub fn validate(&self, skip_rproofs: bool) -> Result<(), Error> {
 		let header = self.store.head_header()?;
@@ -481,28 +523,13 @@ impl Chain {
 		})
 	}
 
-	/// Check if the input has matured sufficiently for the given block height.
-	/// This only applies to inputs spending coinbase outputs.
-	/// An input spending a non-coinbase output will always pass this check.
-	pub fn is_matured(&self, input: &Input, height: u64) -> Result<(), Error> {
-		if input.features.contains(OutputFeatures::COINBASE_OUTPUT) {
-			let mut txhashset = self.txhashset.write().unwrap();
-			let output = OutputIdentifier::from_input(&input);
-			let hash = txhashset.is_unspent(&output)?;
-			let header = self.get_block_header(&input.block_hash())?;
-			input.verify_maturity(hash, &header, height)?;
-		}
-		Ok(())
-	}
-
-	/// Sets the txhashset roots on a brand new block by applying the block on the
-	/// current txhashset state.
+	/// Sets the txhashset roots on a brand new block by applying the block on
+	/// the current txhashset state.
 	pub fn set_txhashset_roots(&self, b: &mut Block, is_fork: bool) -> Result<(), Error> {
 		let mut txhashset = self.txhashset.write().unwrap();
 		let store = self.store.clone();
 
 		let roots = txhashset::extending_readonly(&mut txhashset, |extension| {
-			// apply the block on the txhashset and check the resulting root
 			if is_fork {
 				pipe::rewind_and_apply_fork(b, store, extension)?;
 			}
@@ -810,7 +837,8 @@ impl Chain {
 	}
 
 	/// Verifies the given block header is actually on the current chain.
-	/// Checks the header_by_height index to verify the header is where we say it is
+	/// Checks the header_by_height index to verify the header is where we say
+	/// it is
 	pub fn is_on_current_chain(&self, header: &BlockHeader) -> Result<(), Error> {
 		self.store
 			.is_on_current_chain(header)

chain/src/lib.rs

@@ -45,4 +45,4 @@ pub mod types;
 // Re-export the base interface
 
 pub use chain::{Chain, MAX_ORPHAN_SIZE};
-pub use types::{ChainAdapter, ChainStore, Error, Options, Tip};
+pub use types::{BlockSums, ChainAdapter, ChainStore, Error, Options, Tip};

chain/src/pipe.rs

@@ -20,13 +20,13 @@ use time;
 
 use core::consensus;
 use core::core::hash::{Hash, Hashed};
-use core::core::{Block, BlockHeader};
 use core::core::target::Difficulty;
+use core::core::{Block, BlockHeader};
+use core::global;
 use grin_store;
-use types::*;
 use store;
 use txhashset;
-use core::global;
+use types::*;
 use util::LOGGER;
 
 /// Contextual information required to process a new block and either reject or
@@ -100,7 +100,14 @@ pub fn process_block(b: &Block, mut ctx: BlockContext) -> Result<Option<Tip>, Er
 	// start a chain extension unit of work dependent on the success of the
 	// internal validation and saving operations
 	let result = txhashset::extending(&mut txhashset, |mut extension| {
-		validate_block_via_txhashset(b, &mut ctx, &mut extension)?;
+		// First we rewind the txhashset extension if necessary
+		// to put it into a consistent state for validating the block.
+		// We can skip this step if the previous header is the latest header we saw.
+		if b.header.previous != ctx.head.last_block_h {
+			rewind_and_apply_fork(b, ctx.store.clone(), extension)?;
+		}
+		validate_block_via_txhashset(b, &mut extension)?;
+
 		trace!(
 			LOGGER,
 			"pipe: process_block: {} at {} is valid, save and append.",
@@ -143,8 +150,9 @@ pub fn sync_block_header(
 }
 
 /// Process block header as part of "header first" block propagation.
-/// We validate the header but we do not store it or update header head based on this.
-/// We will update these once we get the block back after requesting it.
+/// We validate the header but we do not store it or update header head based
+/// on this. We will update these once we get the block back after requesting
+/// it.
 pub fn process_block_header(bh: &BlockHeader, mut ctx: BlockContext) -> Result<(), Error> {
 	debug!(
 		LOGGER,
@@ -326,14 +334,10 @@ fn validate_block(b: &Block, ctx: &mut BlockContext) -> Result<(), Error> {
 /// and checking the roots.
 /// Rewind and reapply forked blocks if necessary to put the txhashset extension
 /// in the correct state to accept the block.
-fn validate_block_via_txhashset(
-	b: &Block,
-	ctx: &mut BlockContext,
-	ext: &mut txhashset::Extension,
-) -> Result<(), Error> {
-	if b.header.previous != ctx.head.last_block_h {
-		rewind_and_apply_fork(b, ctx.store.clone(), ext)?;
-	}
+fn validate_block_via_txhashset(b: &Block, ext: &mut txhashset::Extension) -> Result<(), Error> {
+	// First check we are not attempting to spend any coinbase outputs
+	// before they have matured sufficiently.
+	ext.verify_coinbase_maturity(&b.inputs, b.header.height)?;
 
 	// apply the new block to the MMR trees and check the new root hashes
 	ext.apply_block(&b)?;
@@ -515,7 +519,7 @@ pub fn rewind_and_apply_fork(
 
 	let forked_block = store.get_block_header(&current)?;
 
-	debug!(
+	trace!(
 		LOGGER,
 		"rewind_and_apply_fork @ {} [{}], was @ {} [{}]",
 		forked_block.height,
@@ -527,9 +531,10 @@ pub fn rewind_and_apply_fork(
 	// rewind the sum trees up to the forking block
 	ext.rewind(&forked_block)?;
 
-	debug!(
+	trace!(
 		LOGGER,
-		"rewind_and_apply_fork: blocks on fork: {:?}", hashes
+		"rewind_and_apply_fork: blocks on fork: {:?}",
+		hashes,
 	);
 
 	// apply all forked blocks, including this new one

chain/src/txhashset.rs

@@ -28,7 +28,7 @@ use core::consensus::REWARD;
 use core::core::hash::{Hash, Hashed};
 use core::core::pmmr::{self, MerkleProof, PMMR};
 use core::core::{Block, BlockHeader, Committed, Input, Output, OutputFeatures, OutputIdentifier,
-                 TxKernel};
+                 Transaction, TxKernel};
 use core::global;
 use core::ser::{PMMRIndexHashable, PMMRable};
 
@@ -257,14 +257,16 @@ where
 
 		trace!(LOGGER, "Starting new txhashset (readonly) extension.");
 		let mut extension = Extension::new(trees, commit_index);
+		extension.force_rollback();
 		res = inner(&mut extension);
 
 		sizes = extension.sizes();
 	}
 
-	debug!(
+	trace!(
 		LOGGER,
-		"Rollbacking txhashset (readonly) extension. sizes {:?}", sizes
+		"Rollbacking txhashset (readonly) extension. sizes {:?}",
+		sizes
 	);
 
 	trees.output_pmmr_h.backend.discard();
@@ -315,12 +317,12 @@ where
 		}
 		Ok(r) => {
 			if rollback {
-				debug!(LOGGER, "Rollbacking txhashset extension. sizes {:?}", sizes);
+				trace!(LOGGER, "Rollbacking txhashset extension. sizes {:?}", sizes);
 				trees.output_pmmr_h.backend.discard();
 				trees.rproof_pmmr_h.backend.discard();
 				trees.kernel_pmmr_h.backend.discard();
 			} else {
-				debug!(LOGGER, "Committing txhashset extension. sizes {:?}", sizes);
+				trace!(LOGGER, "Committing txhashset extension. sizes {:?}", sizes);
 				trees.output_pmmr_h.backend.sync()?;
 				trees.rproof_pmmr_h.backend.sync()?;
 				trees.kernel_pmmr_h.backend.sync()?;
@@ -402,6 +404,106 @@ impl<'a> Extension<'a> {
 		}
 	}
 
+	/// Apply a "raw" transaction to the txhashset.
+	/// We will never commit a txhashset extension that includes raw txs.
+	/// But we can use this when validating txs in the tx pool.
+	/// If we can add a tx to the tx pool and then successfully add the
+	/// aggregated tx from the tx pool to the current chain state (via a
+	/// txhashset extension) then we know the tx pool is valid (including the
+	/// new tx).
+	pub fn apply_raw_tx(&mut self, tx: &Transaction, height: u64) -> Result<(), Error> {
+		// This should *never* be called on a writeable extension...
+		if !self.rollback {
+			panic!("attempted to apply a raw tx to a writeable txhashset extension");
+		}
+
+		// Checkpoint the MMR positions before we apply the new tx,
+		// anything goes wrong we will rewind to these positions.
+		let output_pos = self.output_pmmr.unpruned_size();
+		let kernel_pos = self.kernel_pmmr.unpruned_size();
+
+		// When applying blocks we can apply the coinbase output first
+		// but we cannot do this here, so we need to apply outputs first.
+		for ref output in &tx.outputs {
+			if let Err(e) = self.apply_output(output) {
+				self.rewind_to_pos(output_pos, kernel_pos, height)?;
+				return Err(e);
+			}
+		}
+
+		for ref input in &tx.inputs {
+			if let Err(e) = self.apply_input(input, height) {
+				self.rewind_to_pos(output_pos, kernel_pos, height)?;
+				return Err(e);
+			}
+		}
+
+		for ref kernel in &tx.kernels {
+			if let Err(e) = self.apply_kernel(kernel) {
+				self.rewind_to_pos(output_pos, kernel_pos, height)?;
+				return Err(e);
+			}
+		}
+
+		Ok(())
+	}
+
+	/// Validate a vector of "raw" transactions against the current chain state.
+	/// We support rewind on a "dirty" txhashset - so we can apply each tx in
+	/// turn, rewinding if any particular tx is not valid and continuing
+	/// through the vec of txs provided. This allows us to efficiently apply
+	/// all the txs, filtering out those that are not valid and returning the
+	/// final vec of txs that were successfully validated against the txhashset.
+	///
+	/// Note: We also pass in a "pre_tx". This tx is applied to and validated
+	/// before we start applying the vec of txs. We use this when validating
+	/// txs in the stempool as we need to account for txs in the txpool as
+	/// well (new_tx + stempool + txpool + txhashset). So we aggregate the
+	/// contents of the txpool into a single aggregated tx and pass it in here
+	/// as the "pre_tx" so we apply it to the txhashset before we start
+	/// validating the stempool txs.
+	/// This is optional and we pass in None when validating the txpool txs
+	/// themselves.
+	///
+	pub fn validate_raw_txs(
+		&mut self,
+		txs: Vec<Transaction>,
+		pre_tx: Option<Transaction>,
+		height: u64,
+	) -> Result<Vec<Transaction>, Error> {
+		let mut height = height;
+		let mut valid_txs = vec![];
+		if let Some(tx) = pre_tx {
+			height += 1;
+			self.apply_raw_tx(&tx, height)?;
+		}
+		for tx in txs {
+			height += 1;
+			if self.apply_raw_tx(&tx, height).is_ok() {
+				valid_txs.push(tx);
+			}
+		}
+		Ok(valid_txs)
+	}
+
+	/// Verify we are not attempting to spend any coinbase outputs
+	/// that have not sufficiently matured.
+	pub fn verify_coinbase_maturity(
+		&mut self,
+		inputs: &Vec<Input>,
+		height: u64,
+	) -> Result<(), Error> {
+		for x in inputs {
+			if x.features.contains(OutputFeatures::COINBASE_OUTPUT) {
+				let header = self.commit_index.get_block_header(&x.block_hash())?;
+				let pos = self.get_output_pos(&x.commitment())?;
+				let out_hash = self.output_pmmr.get_hash(pos).ok_or(Error::OutputNotFound)?;
+				x.verify_maturity(out_hash, &header, height)?;
+			}
+		}
+		Ok(())
+	}
+
 	/// Apply a new set of blocks on top the existing sum trees. Blocks are
 	/// applied in order of the provided Vec. If pruning is enabled, inputs also
 	/// prune MMR data.
@@ -443,8 +545,9 @@ impl<'a> Extension<'a> {
 		Ok(())
 	}
 
+	// Store all new output pos in the index.
+	// Also store any new block_markers.
 	fn save_indexes(&self) -> Result<(), Error> {
-		// store all new output pos in the index
 		for (commit, pos) in &self.new_output_commits {
 			self.commit_index.save_output_pos(commit, *pos)?;
 		}
@@ -463,23 +566,13 @@ impl<'a> Extension<'a> {
 				// check hash from pmmr matches hash from input (or corresponding output)
 				// if not then the input is not being honest about
 				// what it is attempting to spend...
-
 				let read_elem = self.output_pmmr.get_data(pos);
-
-				if output_id_hash != read_hash
-					|| output_id_hash
-						!= read_elem
-							.expect("no output at position")
-							.hash_with_index(pos - 1)
-				{
+				let read_elem_hash = read_elem
+					.expect("no output at pos")
+					.hash_with_index(pos - 1);
+				if output_id_hash != read_hash || output_id_hash != read_elem_hash {
 					return Err(Error::TxHashSetErr(format!("output pmmr hash mismatch")));
 				}
-
-				// check coinbase maturity with the Merkle Proof on the input
-				if input.features.contains(OutputFeatures::COINBASE_OUTPUT) {
-					let header = self.commit_index.get_block_header(&input.block_hash())?;
-					input.verify_maturity(read_hash, &header, height)?;
-				}
 			}
 
 			// Now prune the output_pmmr, rproof_pmmr and their storage.
@@ -502,7 +595,6 @@ impl<'a> Extension<'a> {
 
 	fn apply_output(&mut self, out: &Output) -> Result<(), Error> {
 		let commit = out.commitment();
-
 		if let Ok(pos) = self.get_output_pos(&commit) {
 			// we need to check whether the commitment is in the current MMR view
 			// as well as the index doesn't support rewind and is non-authoritative
@@ -581,28 +673,40 @@ impl<'a> Extension<'a> {
 	pub fn rewind(&mut self, block_header: &BlockHeader) -> Result<(), Error> {
 		let hash = block_header.hash();
 		let height = block_header.height;
-		debug!(LOGGER, "Rewind to header {} @ {}", height, hash);
+		trace!(LOGGER, "Rewind to header {} @ {}", height, hash);
 
 		// rewind our MMRs to the appropriate pos
 		// based on block height and block marker
 		let marker = self.commit_index.get_block_marker(&hash)?;
-		self.rewind_to_marker(height, &marker)?;
+		self.rewind_to_pos(marker.output_pos, marker.kernel_pos, height)?;
+
 		Ok(())
 	}
 
 	/// Rewinds the MMRs to the provided positions, given the output and
 	/// kernel we want to rewind to.
-	fn rewind_to_marker(&mut self, height: u64, marker: &BlockMarker) -> Result<(), Error> {
-		debug!(LOGGER, "Rewind txhashset to {}, {:?}", height, marker);
+	fn rewind_to_pos(
+		&mut self,
+		output_pos: u64,
+		kernel_pos: u64,
+		height: u64,
+	) -> Result<(), Error> {
+		trace!(
+			LOGGER,
+			"Rewind txhashset to output {}, kernel {}, height {}",
+			output_pos,
+			kernel_pos,
+			height
+		);
 
 		self.output_pmmr
-			.rewind(marker.output_pos, height as u32)
+			.rewind(output_pos, height as u32)
 			.map_err(&Error::TxHashSetErr)?;
 		self.rproof_pmmr
-			.rewind(marker.output_pos, height as u32)
+			.rewind(output_pos, height as u32)
 			.map_err(&Error::TxHashSetErr)?;
 		self.kernel_pmmr
-			.rewind(marker.kernel_pos, height as u32)
+			.rewind(kernel_pos, height as u32)
 			.map_err(&Error::TxHashSetErr)?;
 
 		Ok(())

chain/src/types.rs

@@ -99,6 +99,8 @@ pub enum Error {
 	SerErr(ser::Error),
 	/// Error with the txhashset
 	TxHashSetErr(String),
+	/// Tx not valid based on lock_height.
+	TxLockHeight,
 	/// No chain exists and genesis block is required
 	GenesisBlockRequired,
 	/// Error from underlying tx handling