Merge branch 'master' into new-jsonrpc

rpc/src/v1/helpers/mod.rs

@@ -39,5 +39,6 @@ pub use self::requests::{
 };
 pub use self::signing_queue::{
 	ConfirmationsQueue, ConfirmationPromise, ConfirmationResult, SigningQueue, QueueEvent, DefaultAccount,
+	QUEUE_LIMIT as SIGNING_QUEUE_LIMIT,
 };
 pub use self::signer::SignerService;

rpc/src/v1/helpers/signing_queue.rs

@@ -77,7 +77,7 @@ pub enum QueueAddError {
 }
 
 // TODO [todr] to consider: timeout instead of limit?
-const QUEUE_LIMIT: usize = 50;
+pub const QUEUE_LIMIT: usize = 50;
 
 /// A queue of transactions awaiting to be confirmed and signed.
 pub trait SigningQueue: Send + Sync {

rpc/src/v1/impls/light/parity_set.rs

@@ -28,7 +28,7 @@ use util::sha3;
 use jsonrpc_core::Error;
 use v1::helpers::errors;
 use v1::traits::ParitySet;
-use v1::types::{Bytes, H160, H256, U256, ReleaseInfo};
+use v1::types::{Bytes, H160, H256, U256, ReleaseInfo, Transaction};
 
 /// Parity-specific rpc interface for operations altering the settings.
 pub struct ParitySetClient<F> {
@@ -139,4 +139,8 @@ impl<F: Fetch> ParitySet for ParitySetClient<F> {
 	fn execute_upgrade(&self) -> Result<bool, Error> {
 		Err(errors::light_unimplemented(None))
 	}
+
+	fn remove_transaction(&self, _hash: H256) -> Result<Option<Transaction>, Error> {
+		Err(errors::light_unimplemented(None))
+	}
 }

rpc/src/v1/impls/parity_set.rs

@@ -30,15 +30,10 @@ use updater::{Service as UpdateService};
 use jsonrpc_core::Error;
 use v1::helpers::errors;
 use v1::traits::ParitySet;
-use v1::types::{Bytes, H160, H256, U256, ReleaseInfo};
+use v1::types::{Bytes, H160, H256, U256, ReleaseInfo, Transaction};
 
 /// Parity-specific rpc interface for operations altering the settings.
-pub struct ParitySetClient<C, M, U, F=fetch::Client> where
-	C: MiningBlockChainClient,
-	M: MinerService,
-	U: UpdateService,
-	F: Fetch,
-{
+pub struct ParitySetClient<C, M, U, F = fetch::Client> {
 	client: Weak<C>,
 	miner: Weak<M>,
 	updater: Weak<U>,
@@ -46,12 +41,7 @@ pub struct ParitySetClient<C, M, U, F=fetch::Client> where
 	fetch: F,
 }
 
-impl<C, M, U, F> ParitySetClient<C, M, U, F> where
-	C: MiningBlockChainClient,
-	M: MinerService,
-	U: UpdateService,
-	F: Fetch,
-{
+impl<C, M, U, F> ParitySetClient<C, M, U, F> {
 	/// Creates new `ParitySetClient` with given `Fetch`.
 	pub fn new(client: &Arc<C>, miner: &Arc<M>, updater: &Arc<U>, net: &Arc<ManageNetwork>, fetch: F) -> Self {
 		ParitySetClient {
@@ -181,4 +171,12 @@ impl<C, M, U, F> ParitySet for ParitySetClient<C, M, U, F> where
 		let updater = take_weak!(self.updater);
 		Ok(updater.execute_upgrade())
 	}
+
+	fn remove_transaction(&self, hash: H256) -> Result<Option<Transaction>, Error> {
+		let miner = take_weak!(self.miner);
+		let client = take_weak!(self.client);
+		let hash = hash.into();
+
+		Ok(miner.remove_pending_transaction(&*client, &hash).map(Into::into))
+	}
 }

rpc/src/v1/impls/signing.rs

@@ -27,7 +27,7 @@ use jsonrpc_core::Error;
 use v1::helpers::{
 	errors, oneshot,
 	DefaultAccount,
-	SigningQueue, ConfirmationPromise, ConfirmationResult, SignerService
+	SIGNING_QUEUE_LIMIT, SigningQueue, ConfirmationPromise, ConfirmationResult, SignerService
 };
 use v1::helpers::dispatch::{self, Dispatcher};
 use v1::metadata::Metadata;
@@ -42,7 +42,10 @@ use v1::types::{
 	Origin,
 };
 
-const MAX_PENDING_DURATION: u64 = 60 * 60;
+/// After 60s entries that are not queried with `check_request` will get garbage collected.
+const MAX_PENDING_DURATION_SEC: u64 = 60;
+/// Max number of total requests pending and completed, before we start garbage collecting them.
+const MAX_TOTAL_REQUESTS: usize = SIGNING_QUEUE_LIMIT;
 
 enum DispatchResult {
 	Promise(ConfirmationPromise),
@@ -71,6 +74,21 @@ fn handle_dispatch<OnResponse>(res: Result<DispatchResult, Error>, on_response:
 	}
 }
 
+fn collect_garbage(map: &mut TransientHashMap<U256, ConfirmationPromise>) {
+	map.prune();
+	if map.len() > MAX_TOTAL_REQUESTS {
+		// Remove all non-waiting entries.
+		let non_waiting: Vec<_> = map
+			.iter()
+			.filter(|&(_, val)| val.result() != ConfirmationResult::Waiting)
+			.map(|(key, _)| *key)
+			.collect();
+		for k in non_waiting {
+			map.remove(&k);
+		}
+	}
+}
+
 impl<D: Dispatcher + 'static> SigningQueueClient<D> {
 	/// Creates a new signing queue client given shared signing queue.
 	pub fn new(signer: &Arc<SignerService>, dispatcher: D, accounts: &Arc<AccountProvider>) -> Self {
@@ -78,7 +96,7 @@ impl<D: Dispatcher + 'static> SigningQueueClient<D> {
 			signer: Arc::downgrade(signer),
 			accounts: Arc::downgrade(accounts),
 			dispatcher: dispatcher,
-			pending: Arc::new(Mutex::new(TransientHashMap::new(MAX_PENDING_DURATION))),
+			pending: Arc::new(Mutex::new(TransientHashMap::new(MAX_PENDING_DURATION_SEC))),
 		}
 	}
 
@@ -124,7 +142,10 @@ impl<D: Dispatcher + 'static> ParitySigning for SigningQueueClient<D> {
 			DispatchResult::Value(v) => RpcEither::Or(v),
 			DispatchResult::Promise(promise) => {
 				let id = promise.id();
-				pending.lock().insert(id, promise);
+				let mut pending = pending.lock();
+				collect_garbage(&mut pending);
+				pending.insert(id, promise);
+
 				RpcEither::Either(id.into())
 			},
 		})
@@ -138,7 +159,10 @@ impl<D: Dispatcher + 'static> ParitySigning for SigningQueueClient<D> {
 				DispatchResult::Value(v) => RpcEither::Or(v),
 				DispatchResult::Promise(promise) => {
 					let id = promise.id();
-					pending.lock().insert(id, promise);
+					let mut pending = pending.lock();
+					collect_garbage(&mut pending);
+					pending.insert(id, promise);
+
 					RpcEither::Either(id.into())
 				},
 			})
@@ -146,18 +170,15 @@ impl<D: Dispatcher + 'static> ParitySigning for SigningQueueClient<D> {
 	}
 
 	fn check_request(&self, id: RpcU256) -> Result<Option<RpcConfirmationResponse>, Error> {
-		let mut pending = self.pending.lock();
 		let id: U256 = id.into();
-		let res = match pending.get(&id) {
+		match self.pending.lock().get(&id) {
 			Some(ref promise) => match promise.result() {
-				ConfirmationResult::Waiting => { return Ok(None); }
+				ConfirmationResult::Waiting => Ok(None),
 				ConfirmationResult::Rejected => Err(errors::request_rejected()),
 				ConfirmationResult::Confirmed(rpc_response) => rpc_response.map(Some),
 			},
-			_ => { return Err(errors::request_not_found()); }
-		};
-		pending.remove(&id);
-		res
+			_ => Err(errors::request_not_found()),
+		}
 	}
 
 	fn decrypt_message(&self, meta: Metadata, address: RpcH160, data: RpcBytes) -> BoxFuture<RpcBytes, Error> {

rpc/src/v1/tests/helpers/miner_service.rs

@@ -221,6 +221,10 @@ impl MinerService for TestMinerService {
 		self.pending_transactions.lock().get(hash).cloned().map(Into::into)
 	}
 
+	fn remove_pending_transaction(&self, _chain: &MiningBlockChainClient, hash: &H256) -> Option<PendingTransaction> {
+		self.pending_transactions.lock().remove(hash).map(Into::into)
+	}
+
 	fn pending_transactions(&self) -> Vec<PendingTransaction> {
 		self.pending_transactions.lock().values().cloned().map(Into::into).collect()
 	}

rpc/src/v1/tests/mocked/parity_set.rs

@@ -204,3 +204,31 @@ fn rpc_parity_set_hash_content() {
 	assert_eq!(io.handle_request_sync(request), Some(response.to_owned()));
 }
 
+#[test]
+fn rpc_parity_remove_transaction() {
+	use ethcore::transaction::{Transaction, Action};
+
+	let miner = miner_service();
+	let client = client_service();
+	let network = network_service();
+	let updater = updater_service();
+	let mut io = IoHandler::new();
+	io.extend_with(parity_set_client(&client, &miner, &updater, &network).to_delegate());
+
+	let tx = Transaction {
+		nonce: 1.into(),
+		gas_price: 0x9184e72a000u64.into(),
+		gas: 0x76c0.into(),
+		action: Action::Call(5.into()),
+		value: 0x9184e72au64.into(),
+		data: vec![]
+	};
+	let signed = tx.fake_sign(2.into());
+	let hash = signed.hash();
+
+	let request = r#"{"jsonrpc": "2.0", "method": "parity_removeTransaction", "params":[""#.to_owned() + &format!("0x{:?}", hash) + r#""], "id": 1}"#;
+	let response = r#"{"jsonrpc":"2.0","result":{"blockHash":null,"blockNumber":null,"condition":null,"creates":null,"from":"0x0000000000000000000000000000000000000002","gas":"0x76c0","gasPrice":"0x9184e72a000","hash":"0x0072c69d780cdfbfc02fed5c7d184151f9a166971d045e55e27695aaa5bcb55e","input":"0x","networkId":null,"nonce":"0x1","publicKey":"0x00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000","r":"0x0","raw":"0xe9018609184e72a0008276c0940000000000000000000000000000000000000005849184e72a80808080","s":"0x0","standardV":"0x4","to":"0x0000000000000000000000000000000000000005","transactionIndex":null,"v":"0x0","value":"0x9184e72a"},"id":1}"#;
+
+	miner.pending_transactions.lock().insert(hash, signed);
+	assert_eq!(io.handle_request_sync(&request), Some(response.to_owned()));
+}

rpc/src/v1/traits/parity_set.rs

@@ -19,7 +19,7 @@
 use jsonrpc_core::Error;
 use futures::BoxFuture;
 
-use v1::types::{Bytes, H160, H256, U256, ReleaseInfo};
+use v1::types::{Bytes, H160, H256, U256, ReleaseInfo, Transaction};
 
 build_rpc_trait! {
 	/// Parity-specific rpc interface for operations altering the settings.
@@ -103,5 +103,14 @@ build_rpc_trait! {
 		/// Execute a release which is ready according to upgrade_ready().
 		#[rpc(name = "parity_executeUpgrade")]
 		fn execute_upgrade(&self) -> Result<bool, Error>;
+
+		/// Removes transaction from transaction queue.
+		/// Makes sense only for transactions that were not propagated to other peers yet
+		/// like scheduled transactions or transactions in future.
+		/// It might also work for some local transactions with to low gas price
+		/// or excessive gas limit that are not accepted by other peers whp.
+		/// Returns `true` when transaction was removed, `false` if it was not found.
+		#[rpc(name = "parity_removeTransaction")]
+		fn remove_transaction(&self, H256) -> Result<Option<Transaction>, Error>;
 	}
 }