Private packets verification and queue refactoring

ethcore/private-tx/Cargo.toml

@@ -22,6 +22,7 @@ ethjson = { path = "../../json" }
 ethkey = { path = "../../ethkey" }
 fetch = { path = "../../util/fetch" }
 futures = "0.1"
+heapsize = "0.4"
 keccak-hash = { git = "https://github.com/paritytech/parity-common" }
 log = "0.3"
 parking_lot = "0.6"
@@ -35,6 +36,7 @@ serde = "1.0"
 serde_derive = "1.0"
 serde_json = "1.0"
 tiny-keccak = "1.4"
+transaction-pool = { path = "../../transaction-pool" }
 url = "1"
 
 [dev-dependencies]

ethcore/private-tx/src/error.rs

@@ -21,12 +21,14 @@ use ethcore::account_provider::SignError;
 use ethcore::error::{Error as EthcoreError, ExecutionError};
 use transaction::Error as TransactionError;
 use ethkey::Error as KeyError;
+use txpool::Error as TxPoolError;
 
 error_chain! {
 	foreign_links {
 		Io(::std::io::Error) #[doc = "Error concerning the Rust standard library's IO subsystem."];
 		Decoder(DecoderError) #[doc = "RLP decoding error."];
 		Trie(TrieError) #[doc = "Error concerning TrieDBs."];
+		Txpool(TxPoolError) #[doc = "Tx pool error."];
 	}
 
 	errors {

ethcore/private-tx/src/messages.rs

@@ -28,12 +28,28 @@ pub struct PrivateTransaction {
 	pub encrypted: Bytes,
 	/// Address of the contract
 	pub contract: Address,
+	/// Hash
+	hash: H256,
 }
 
 impl PrivateTransaction {
-	/// Compute hash on private transaction
+	/// Constructor
+	pub fn new(encrypted: Bytes, contract: Address) -> Self {
+		PrivateTransaction {
+			encrypted,
+			contract,
+			hash: 0.into(),
+		}.compute_hash()
+	}
+
+	fn compute_hash(mut self) -> PrivateTransaction {
+		self.hash = keccak(&*self.rlp_bytes());
+		self
+	}
+
+	/// Hash of the private transaction
 	pub fn hash(&self) -> H256 {
-		keccak(&*self.rlp_bytes())
+		self.hash
 	}
 }
 
@@ -49,6 +65,8 @@ pub struct SignedPrivateTransaction {
 	r: U256,
 	/// The S field of the signature
 	s: U256,
+	/// Hash
+	hash: H256,
 }
 
 impl SignedPrivateTransaction {
@@ -59,7 +77,13 @@ impl SignedPrivateTransaction {
 			r: sig.r().into(),
 			s: sig.s().into(),
 			v: add_chain_replay_protection(sig.v() as u64, chain_id),
-		}
+			hash: 0.into(),
+		}.compute_hash()
+	}
+
+	fn compute_hash(mut self) -> SignedPrivateTransaction {
+		self.hash = keccak(&*self.rlp_bytes());
+		self
 	}
 
 	pub fn standard_v(&self) -> u8 { check_replay_protection(self.v) }
@@ -73,4 +97,9 @@ impl SignedPrivateTransaction {
 	pub fn private_transaction_hash(&self) -> H256 {
 		self.private_transaction_hash
 	}
+
+	/// Own hash
+	pub fn hash(&self) -> H256 {
+		self.hash
+	}
 }

ethcore/private-tx/src/private_transactions.rs

@@ -15,128 +15,188 @@
 // along with Parity.  If not, see <http://www.gnu.org/licenses/>.
 
 use std::sync::Arc;
-use std::collections::{HashMap, HashSet};
+use std::cmp;
+use std::collections::HashMap;
+use std::collections::hash_map::Entry;
 
 use bytes::Bytes;
 use ethcore_miner::pool;
 use ethereum_types::{H256, U256, Address};
+use heapsize::HeapSizeOf;
 use ethkey::Signature;
+use messages::PrivateTransaction;
+use parking_lot::RwLock;
 use transaction::{UnverifiedTransaction, SignedTransaction};
-
+use txpool;
+use txpool::{VerifiedTransaction, Verifier};
 use error::{Error, ErrorKind};
 
+type Pool = txpool::Pool<VerifiedPrivateTransaction, pool::scoring::NonceAndGasPrice>;
+
 /// Maximum length for private transactions queues.
 const MAX_QUEUE_LEN: usize = 8312;
 
 /// Desriptor for private transaction stored in queue for verification
-#[derive(Default, Debug, Clone, PartialEq, Eq)]
-pub struct PrivateTransactionDesc {
-	/// Hash of the private transaction
-	pub private_hash: H256,
-	/// Contract's address used in private transaction
-	pub contract: Address,
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct VerifiedPrivateTransaction {
+	/// Original private transaction
+	pub private_transaction: PrivateTransaction,
 	/// Address that should be used for verification
-	pub validator_account: Address,
+	pub validator_account: Option<Address>,
+	/// Resulted verified
+	pub transaction: SignedTransaction,
+	/// Original transaction's hash
+	pub transaction_hash: H256,
+	/// Original transaction's sender
+	pub transaction_sender: Address,
+}
+
+impl txpool::VerifiedTransaction for VerifiedPrivateTransaction {
+	type Hash = H256;
+	type Sender = Address;
+
+	fn hash(&self) -> &H256 {
+		&self.transaction_hash
+	}
+
+	fn mem_usage(&self) -> usize {
+		self.transaction.heap_size_of_children()
+	}
+
+	fn sender(&self) -> &Address {
+		&self.transaction_sender
+	}
+}
+
+impl pool::ScoredTransaction for VerifiedPrivateTransaction {
+	fn priority(&self) -> pool::Priority {
+		pool::Priority::Regular
+	}
+
+	/// Gets transaction gas price.
+	fn gas_price(&self) -> U256 {
+		self.transaction.gas_price
+	}
+
+	/// Gets transaction nonce.
+	fn nonce(&self) -> U256 {
+		self.transaction.nonce
+	}
+}
+
+/// Checks readiness of transactions by comparing the nonce to state nonce.
+/// Guarantees only one transaction per sender
+#[derive(Debug)]
+pub struct PrivateReadyState<C> {
+	nonces: HashMap<Address, U256>,
+	state: C,
+}
+
+impl<C> PrivateReadyState<C> {
+	/// Create new State checker, given client interface.
+	pub fn new(
+		state: C,
+	) -> Self {
+		PrivateReadyState {
+			nonces: Default::default(),
+			state,
+		}
+	}
+}
+
+impl<C: pool::client::NonceClient> txpool::Ready<VerifiedPrivateTransaction> for PrivateReadyState<C> {
+	fn is_ready(&mut self, tx: &VerifiedPrivateTransaction) -> txpool::Readiness {
+		let sender = tx.sender();
+		let state = &self.state;
+		let state_nonce = state.account_nonce(sender);
+		match self.nonces.entry(*sender) {
+			Entry::Vacant(entry) => {
+				let nonce = entry.insert(state_nonce);
+				match tx.transaction.nonce.cmp(nonce) {
+					cmp::Ordering::Greater => txpool::Readiness::Future,
+					cmp::Ordering::Less => txpool::Readiness::Stale,
+					cmp::Ordering::Equal => {
+						*nonce = *nonce + 1.into();
+						txpool::Readiness::Ready
+					},
+				}
+			}
+			Entry::Occupied(_) => {
+				txpool::Readiness::Future
+			}
+		}
+	}
 }
 
 /// Storage for private transactions for verification
 pub struct VerificationStore {
-	/// Descriptors for private transactions in queue for verification with key - hash of the original transaction
-	descriptors: HashMap<H256, PrivateTransactionDesc>,
-	/// Queue with transactions for verification
-	///
-	/// TODO [ToDr] Might actually be better to use `txpool` directly and:
-	/// 1. Store descriptors inside `VerifiedTransaction`
-	/// 2. Use custom `ready` implementation to only fetch one transaction per sender.
-	/// 3. Get rid of passing dummy `block_number` and `timestamp`
-	transactions: pool::TransactionQueue,
+	verification_pool: RwLock<Pool>,
+	verification_options: pool::verifier::Options,
 }
 
 impl Default for VerificationStore {
 	fn default() -> Self {
 		VerificationStore {
-			descriptors: Default::default(),
-			transactions: pool::TransactionQueue::new(
-				pool::Options {
-					max_count: MAX_QUEUE_LEN,
-					max_per_sender: MAX_QUEUE_LEN / 10,
-					max_mem_usage: 8 * 1024 * 1024,
-				},
-				pool::verifier::Options {
-					// TODO [ToDr] This should probably be based on some real values?
-					minimal_gas_price: 0.into(),
-					block_gas_limit: 8_000_000.into(),
-					tx_gas_limit: U256::max_value(),
-				},
-				pool::PrioritizationStrategy::GasPriceOnly,
-			)
+			verification_pool: RwLock::new(
+				txpool::Pool::new(
+					txpool::NoopListener,
+					pool::scoring::NonceAndGasPrice(pool::PrioritizationStrategy::GasPriceOnly),
+					pool::Options {
+						max_count: MAX_QUEUE_LEN,
+						max_per_sender: MAX_QUEUE_LEN / 10,
+						max_mem_usage: 8 * 1024 * 1024,
+					},
+				)
+			),
+			verification_options: pool::verifier::Options {
+				// TODO [ToDr] This should probably be based on some real values?
+				minimal_gas_price: 0.into(),
+				block_gas_limit: 8_000_000.into(),
+				tx_gas_limit: U256::max_value(),
+			},
 		}
 	}
 }
 
 impl VerificationStore {
 	/// Adds private transaction for verification into the store
 	pub fn add_transaction<C: pool::client::Client>(
-		&mut self,
+		&self,
 		transaction: UnverifiedTransaction,
-		contract: Address,
-		validator_account: Address,
-		private_hash: H256,
+		validator_account: Option<Address>,
+		private_transaction: PrivateTransaction,
 		client: C,
 	) -> Result<(), Error> {
-		if self.descriptors.len() > MAX_QUEUE_LEN {
-			bail!(ErrorKind::QueueIsFull);
-		}
 
-		let transaction_hash = transaction.hash();
-		if self.descriptors.get(&transaction_hash).is_some() {
-			bail!(ErrorKind::PrivateTransactionAlreadyImported);
-		}
-
-		let results = self.transactions.import(
-			client,
-			vec![pool::verifier::Transaction::Unverified(transaction)],
-		);
-
-		// Verify that transaction was imported
-		results.into_iter()
-			.next()
-			.expect("One transaction inserted; one result returned; qed")?;
-
-		self.descriptors.insert(transaction_hash, PrivateTransactionDesc {
-			private_hash,
-			contract,
+		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 verified_tx = verifier.verify_transaction(pool::verifier::Transaction::Unverified(transaction.clone()))?;
+		let signed_tx: SignedTransaction = verified_tx.signed().clone();
+		let signed_hash = signed_tx.hash();
+		let signed_sender = signed_tx.sender();
+		let verified = VerifiedPrivateTransaction {
+			private_transaction,
 			validator_account,
-		});
-
+			transaction: signed_tx,
+			transaction_hash: signed_hash,
+			transaction_sender: signed_sender,
+		};
+		let mut pool = self.verification_pool.write();
+		pool.import(verified)?;
 		Ok(())
 	}
 
-	/// Returns transactions ready for verification
+	/// Drains transactions ready for verification from the pool
 	/// Returns only one transaction per sender because several cannot be verified in a row without verification from other peers
-	pub fn ready_transactions<C: pool::client::NonceClient>(&self, client: C) -> Vec<Arc<pool::VerifiedTransaction>> {
-		// We never store PendingTransactions and we don't use internal cache,
-		// so we don't need to provide real block number of timestamp here
-		let block_number = 0;
-		let timestamp = 0;
-		let nonce_cap = None;
-
-		self.transactions.collect_pending(client, block_number, timestamp, nonce_cap, |transactions| {
-			// take only one transaction per sender
-			let mut senders = HashSet::with_capacity(self.descriptors.len());
-			transactions.filter(move |tx| senders.insert(tx.signed().sender())).collect()
-		})
-	}
-
-	/// Returns descriptor of the corresponding private transaction
-	pub fn private_transaction_descriptor(&self, transaction_hash: &H256) -> Result<&PrivateTransactionDesc, Error> {
-		self.descriptors.get(transaction_hash).ok_or(ErrorKind::PrivateTransactionNotFound.into())
-	}
-
-	/// Remove transaction from the queue for verification
-	pub fn remove_private_transaction(&mut self, transaction_hash: &H256) {
-		self.descriptors.remove(transaction_hash);
-		self.transactions.remove(&[*transaction_hash], true);
+	pub fn drain<C: pool::client::NonceClient>(&self, client: C) -> Vec<Arc<VerifiedPrivateTransaction>> {
+		let ready = PrivateReadyState::new(client);
+		let transactions: Vec<_> = self.verification_pool.read().pending(ready).collect();
+		let mut pool = self.verification_pool.write();
+		for tx in &transactions {
+			pool.remove(tx.hash(), true);
+		}
+		transactions
 	}
 }