Fixing transaction queue (#2392)

ethcore/src/block.rs

@@ -411,7 +411,7 @@ impl ClosedBlock {
 	pub fn hash(&self) -> H256 { self.header().rlp_sha3(Seal::Without) }
 
 	/// Turn this into a `LockedBlock`, unable to be reopened again.
-	pub fn lock(mut self) -> LockedBlock {
+	pub fn lock(self) -> LockedBlock {
 		LockedBlock {
 			block: self.block,
 			uncle_bytes: self.uncle_bytes,

ethcore/src/blockchain/blockchain.rs

@@ -1389,7 +1389,6 @@ mod tests {
 			.generate(&mut fork_finalizer).unwrap();
 
 		let b1a_hash = BlockView::new(&b1a).header_view().sha3();
-		let b1b_hash = BlockView::new(&b1b).header_view().sha3();
 		let b2_hash = BlockView::new(&b2).header_view().sha3();
 
 		let t1_hash = t1.hash();

ethcore/src/miner/transaction_queue.rs

@@ -714,7 +714,10 @@ impl TransactionQueue {
 			let order = self.current.drop(sender, &k).expect("iterating over a collection that has been retrieved above;
 															 qed");
 			if k >= current_nonce {
-				self.future.insert(*sender, k, order.update_height(k, current_nonce));
+				let order = order.update_height(k, current_nonce);
+				if let Some(old) = self.future.insert(*sender, k, order.clone()) {
+					Self::replace_orders(*sender, k, old, order, &mut self.future, &mut self.by_hash);
+				}
 			} else {
 				trace!(target: "txqueue", "Removing old transaction: {:?} (nonce: {} < {})", order.hash, k, current_nonce);
 				self.by_hash.remove(&order.hash).expect("All transactions in `future` are also in `by_hash`");
@@ -779,7 +782,9 @@ impl TransactionQueue {
 				self.future.by_gas_price.remove(&order.gas_price, &order.hash);
 				// Put to current
 				let order = order.update_height(current_nonce, first_nonce);
-				self.current.insert(address, current_nonce, order);
+				if let Some(old) = self.current.insert(address, current_nonce, order.clone()) {
+					Self::replace_orders(address, current_nonce, old, order, &mut self.current, &mut self.by_hash);
+				}
 				update_last_nonce_to = Some(current_nonce);
 				current_nonce = current_nonce + U256::one();
 			}
@@ -813,47 +818,49 @@ impl TransactionQueue {
 		let nonce = tx.nonce();
 		let hash = tx.hash();
 
-		let next_nonce = self.last_nonces
-			.get(&address)
-			.cloned()
-			.map_or(state_nonce, |n| n + U256::one());
-
 		// The transaction might be old, let's check that.
 		// This has to be the first test, otherwise calculating
 		// nonce height would result in overflow.
 		if nonce < state_nonce {
 			// Droping transaction
-			trace!(target: "txqueue", "Dropping old transaction: {:?} (nonce: {} < {})", tx.hash(), nonce, next_nonce);
+			trace!(target: "txqueue", "Dropping old transaction: {:?} (nonce: {} < {})", tx.hash(), nonce, state_nonce);
 			return Err(TransactionError::Old);
-		} else if nonce > next_nonce {
+		}
+
+		// Update nonces of transactions in future (remove old transactions)
+		self.update_future(&address, state_nonce);
+		// State nonce could be updated. Maybe there are some more items waiting in future?
+		self.move_matching_future_to_current(address, state_nonce, state_nonce);
+		// Check the next expected nonce (might be updated by move above)
+		let next_nonce = self.last_nonces
+			.get(&address)
+			.cloned()
+			.map_or(state_nonce, |n| n + U256::one());
+
+		// Future transaction
+		if nonce > next_nonce {
 			// We have a gap - put to future.
-			// Update nonces of transactions in future (remove old transactions)
-			self.update_future(&address, state_nonce);
 			// Insert transaction (or replace old one with lower gas price)
 			try!(check_too_cheap(Self::replace_transaction(tx, state_nonce, &mut self.future, &mut self.by_hash)));
-			// Return an error if this transaction is not imported because of limit.
+			// Enforce limit in Future
-			try!(check_if_removed(&address, &nonce, self.future.enforce_limit(&mut self.by_hash)));
+			let removed = self.future.enforce_limit(&mut self.by_hash);
+			// Return an error if this transaction was not imported because of limit.
+			try!(check_if_removed(&address, &nonce, removed));
 
 			debug!(target: "txqueue", "Importing transaction to future: {:?}", hash);
 			debug!(target: "txqueue", "status: {:?}", self.status());
 			return Ok(TransactionImportResult::Future);
 		}
+
+		// We might have filled a gap - move some more transactions from future
+		self.move_matching_future_to_current(address, nonce, state_nonce);
+		self.move_matching_future_to_current(address, nonce + U256::one(), state_nonce);
+
+		// Replace transaction if any
 		try!(check_too_cheap(Self::replace_transaction(tx, state_nonce, &mut self.current, &mut self.by_hash)));
 		// Keep track of highest nonce stored in current
 		let new_max = self.last_nonces.get(&address).map_or(nonce, |n| cmp::max(nonce, *n));
 		self.last_nonces.insert(address, new_max);
-		// Update nonces of transactions in future
-		self.update_future(&address, state_nonce);
-		// Maybe there are some more items waiting in future?
-		self.move_matching_future_to_current(address, nonce + U256::one(), state_nonce);
-		// There might be exactly the same transaction waiting in future
-		// same (sender, nonce), but above function would not move it.
-		if let Some(order) = self.future.drop(&address, &nonce) {
-			// Let's insert that transaction to current (if it has higher gas_price)
-			let future_tx = self.by_hash.remove(&order.hash).expect("All transactions in `future` are always in `by_hash`.");
-			// if transaction in `current` (then one we are importing) is replaced it means that it has to low gas_price
-			try!(check_too_cheap(!Self::replace_transaction(future_tx, state_nonce, &mut self.current, &mut self.by_hash)));
-		}
 
 		// Also enforce the limit
 		let removed = self.current.enforce_limit(&mut self.by_hash);
@@ -898,24 +905,28 @@ impl TransactionQueue {
 
 
 		if let Some(old) = set.insert(address, nonce, order.clone()) {
-			// There was already transaction in queue. Let's check which one should stay
+			Self::replace_orders(address, nonce, old, order, set, by_hash)
-			let old_fee = old.gas_price;
-			let new_fee = order.gas_price;
-			if old_fee.cmp(&new_fee) == Ordering::Greater {
-				// Put back old transaction since it has greater priority (higher gas_price)
-				set.insert(address, nonce, old);
-				// and remove new one
-				by_hash.remove(&hash).expect("The hash has been just inserted and no other line is altering `by_hash`.");
-				false
-			} else {
-				// Make sure we remove old transaction entirely
-				by_hash.remove(&old.hash).expect("The hash is coming from `future` so it has to be in `by_hash`.");
-				true
-			}
 		} else {
 			true
 		}
 	}
+
+	fn replace_orders(address: Address, nonce: U256, old: TransactionOrder, order: TransactionOrder, set: &mut TransactionSet, by_hash: &mut HashMap<H256, VerifiedTransaction>) -> bool {
+		// There was already transaction in queue. Let's check which one should stay
+		let old_fee = old.gas_price;
+		let new_fee = order.gas_price;
+		if old_fee.cmp(&new_fee) == Ordering::Greater {
+			// Put back old transaction since it has greater priority (higher gas_price)
+			set.insert(address, nonce, old);
+			// and remove new one
+			by_hash.remove(&order.hash).expect("The hash has been just inserted and no other line is altering `by_hash`.");
+			false
+		} else {
+			// Make sure we remove old transaction entirely
+			by_hash.remove(&old.hash).expect("The hash is coming from `future` so it has to be in `by_hash`.");
+			true
+		}
+	}
 }
 
 fn check_too_cheap(is_in: bool) -> Result<(), TransactionError> {
@@ -1210,6 +1221,31 @@ mod test {
 		assert_eq!(txq.top_transactions()[0], tx2);
 	}
 
+	#[test]
+	fn should_move_all_transactions_from_future() {
+		// given
+		let mut txq = TransactionQueue::new();
+		let (tx, tx2) = new_tx_pair_default(1.into(), 1.into());
+		let prev_nonce = |a: &Address| AccountDetails{ nonce: default_account_details(a).nonce - U256::one(), balance:
+			!U256::zero() };
+
+		// First insert one transaction to future
+		let res = txq.add(tx.clone(), &prev_nonce, TransactionOrigin::External);
+		assert_eq!(res.unwrap(), TransactionImportResult::Future);
+		assert_eq!(txq.status().future, 1);
+
+		// now import second transaction to current
+		let res = txq.add(tx2.clone(), &default_account_details, TransactionOrigin::External);
+
+		// then
+		assert_eq!(res.unwrap(), TransactionImportResult::Current);
+		assert_eq!(txq.status().pending, 2);
+		assert_eq!(txq.status().future, 0);
+		assert_eq!(txq.current.by_priority.len(), 2);
+		assert_eq!(txq.current.by_address.len(), 2);
+		assert_eq!(txq.top_transactions()[0], tx);
+		assert_eq!(txq.top_transactions()[1], tx2);
+	}
 
 	#[test]
 	fn should_import_tx() {