tx-pool: check transaction readiness before replacing

ethcore/private-tx/Cargo.toml

@@ -36,7 +36,7 @@ serde = "1.0"
 serde_derive = "1.0"
 serde_json = "1.0"
 tiny-keccak = "1.4"
-transaction-pool = "1.13.2"
+transaction-pool = "2.0"
 url = "1"
 
 [dev-dependencies]

ethcore/private-tx/src/error.rs

@@ -23,7 +23,10 @@ use ethcore::error::{Error as EthcoreError, ExecutionError};
 use types::transaction::Error as TransactionError;
 use ethkey::Error as KeyError;
 use ethkey::crypto::Error as CryptoError;
-use txpool::{Error as TxPoolError};
+use txpool::VerifiedTransaction;
+use private_transactions::VerifiedPrivateTransaction;
+
+type TxPoolError = txpool::Error<<VerifiedPrivateTransaction as VerifiedTransaction>::Hash>;
 
 #[derive(Debug, Display)]
 pub enum Error {

ethcore/private-tx/src/private_transactions.rs

@@ -154,7 +154,7 @@ impl Default for VerificationStore {
 
 impl VerificationStore {
 	/// Adds private transaction for verification into the store
-	pub fn add_transaction<C: pool::client::Client>(
+	pub fn add_transaction<C: pool::client::Client + pool::client::NonceClient + Clone>(
 		&self,
 		transaction: UnverifiedTransaction,
 		validator_account: Option<Address>,
@@ -164,7 +164,7 @@ impl VerificationStore {
 
 		let options = self.verification_options.clone();
 		// Use pool's verifying pipeline for original transaction's verification
-		let verifier = pool::verifier::Verifier::new(client, options, Default::default(), None);
+		let verifier = pool::verifier::Verifier::new(client.clone(), options, Default::default(), None);
 		let unverified = pool::verifier::Transaction::Unverified(transaction);
 		let verified_tx = verifier.verify_transaction(unverified)?;
 		let signed_tx: SignedTransaction = verified_tx.signed().clone();
@@ -177,8 +177,9 @@ impl VerificationStore {
 			transaction_hash: signed_hash,
 			transaction_sender: signed_sender,
 		};
-		let mut pool = self.verification_pool.write();
-		pool.import(verified)?;
+		let replace = pool::replace::ReplaceByScoreAndReadiness::new(
+			self.verification_pool.read().scoring().clone(), client);
+		self.verification_pool.write().import(verified, &replace)?;
 		Ok(())
 	}
 

miner/Cargo.toml

@@ -32,7 +32,7 @@ parking_lot = "0.7"
 price-info = { path = "./price-info", optional = true }
 rlp = { version = "0.3.0", features = ["ethereum"] }
 trace-time = "0.1"
-transaction-pool = "1.13"
+transaction-pool = "2.0"
 
 [dev-dependencies]
 env_logger = "0.5"

miner/src/pool/listener.rs

@@ -92,7 +92,7 @@ impl txpool::Listener<Transaction> for Logger {
 		}
 	}
 
-	fn rejected(&mut self, _tx: &Arc<Transaction>, reason: &txpool::ErrorKind) {
+	fn rejected<H: fmt::Debug + fmt::LowerHex>(&mut self, _tx: &Arc<Transaction>, reason: &txpool::Error<H>) {
 		trace!(target: "txqueue", "Rejected {}.", reason);
 	}
 

miner/src/pool/local_transactions.rs

@@ -171,7 +171,7 @@ impl txpool::Listener<Transaction> for LocalTransactionsList {
 		}
 	}
 
-	fn rejected(&mut self, tx: &Arc<Transaction>, reason: &txpool::ErrorKind) {
+	fn rejected<H: fmt::Debug + fmt::LowerHex>(&mut self, tx: &Arc<Transaction>, reason: &txpool::Error<H>) {
 		if !tx.priority().is_local() {
 			return;
 		}

miner/src/pool/mod.rs

@@ -27,6 +27,7 @@ mod ready;
 
 pub mod client;
 pub mod local_transactions;
+pub mod replace;
 pub mod scoring;
 pub mod verifier;
 
@@ -121,7 +122,7 @@ pub trait ScoredTransaction {
 }
 
 /// Verified transaction stored in the pool.
-#[derive(Debug, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct VerifiedTransaction {
 	transaction: transaction::PendingTransaction,
 	// TODO [ToDr] hash and sender should go directly from the transaction

miner/src/pool/queue.rs

@@ -27,7 +27,7 @@ use txpool::{self, Verifier};
 use types::transaction;
 
 use pool::{
-	self, scoring, verifier, client, ready, listener,
+	self, replace, scoring, verifier, client, ready, listener,
 	PrioritizationStrategy, PendingOrdering, PendingSettings,
 };
 use pool::local_transactions::LocalTransactionsList;
@@ -240,7 +240,7 @@ impl TransactionQueue {
 	///
 	/// Given blockchain and state access (Client)
 	/// verifies and imports transactions to the pool.
-	pub fn import<C: client::Client>(
+	pub fn import<C: client::Client + client::NonceClient + Clone>(
 		&self,
 		client: C,
 		transactions: Vec<verifier::Transaction>,
@@ -263,12 +263,14 @@ impl TransactionQueue {
 		};
 
 		let verifier = verifier::Verifier::new(
-			client,
+			client.clone(),
 			options,
 			self.insertion_id.clone(),
 			transaction_to_replace,
 		);
 
+		let mut replace = replace::ReplaceByScoreAndReadiness::new(self.pool.read().scoring().clone(), client);
+
 		let results = transactions
 			.into_iter()
 			.map(|transaction| {
@@ -286,7 +288,7 @@ impl TransactionQueue {
 				let imported = verifier
 					.verify_transaction(transaction)
 					.and_then(|verified| {
-						self.pool.write().import(verified).map_err(convert_error)
+						self.pool.write().import(verified, &mut replace).map_err(convert_error)
 					});
 
 				match imported {
@@ -579,17 +581,13 @@ impl TransactionQueue {
 	}
 }
 
-fn convert_error(err: txpool::Error) -> transaction::Error {
-	use self::txpool::ErrorKind;
-
-	match *err.kind() {
-		ErrorKind::AlreadyImported(..) => transaction::Error::AlreadyImported,
-		ErrorKind::TooCheapToEnter(..) => transaction::Error::LimitReached,
-		ErrorKind::TooCheapToReplace(..) => transaction::Error::TooCheapToReplace,
-		ref e => {
-			warn!(target: "txqueue", "Unknown import error: {:?}", e);
-			transaction::Error::NotAllowed
-		},
+fn convert_error<H: fmt::Debug + fmt::LowerHex>(err: txpool::Error<H>) -> transaction::Error {
+	use self::txpool::Error;
+
+	match err {
+		Error::AlreadyImported(..) => transaction::Error::AlreadyImported,
+		Error::TooCheapToEnter(..) => transaction::Error::LimitReached,
+		Error::TooCheapToReplace(..) => transaction::Error::TooCheapToReplace
 	}
 }