grassrootseconomics
/
openethereum
5
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Enable pending block when there is local transaction pending. (#1005) 

* Enabling sealing while importing own transaction

* Fixing import transaction deadlock

* Checking if there are local transactions in queue

* Updating hyper-mio

* Switching to rust-url#1.0.0
This commit is contained in:
Tomasz Drwięga 2016-04-28 17:36:53 +02:00 committed by Gav Wood
parent d238b5e578
commit ea669ac6b6
5 changed files with 163 additions and 113 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
miner/src/lib.rs
View File

@ -108,7 +108,7 @@ pub trait MinerService : Send + Sync {
where T: Fn(&Address) -> AccountDetails;
/// Imports own (node owner) transaction to queue.
fn import_own_transaction<T>(&self, transaction: SignedTransaction, fetch_account: T) ->
fn import_own_transaction<T>(&self, chain: &BlockChainClient, transaction: SignedTransaction, fetch_account: T) ->
Result<TransactionImportResult, Error>
where T: Fn(&Address) -> AccountDetails;

93
miner/src/miner.rs
View File

@ -154,6 +154,26 @@ impl Miner {
let mut queue = self.transaction_queue.lock().unwrap();
queue.set_gas_limit(gas_limit);
}
/// Returns true if we had to prepare new pending block
fn enable_and_prepare_sealing(&self, chain: &BlockChainClient) -> bool {
trace!(target: "miner", "enable_and_prepare_sealing: entering");
let have_work = self.sealing_work.lock().unwrap().peek_last_ref().is_some();
trace!(target: "miner", "enable_and_prepare_sealing: have_work={}", have_work);
if !have_work {
self.sealing_enabled.store(true, atomic::Ordering::Relaxed);
self.prepare_sealing(chain);
}
let mut sealing_block_last_request = self.sealing_block_last_request.lock().unwrap();
let best_number = chain.chain_info().best_block_number;
if *sealing_block_last_request != best_number {
trace!(target: "miner", "enable_and_prepare_sealing: Miner received request (was {}, now {}) - waking up.", *sealing_block_last_request, best_number);
*sealing_block_last_request = best_number;
}
// Return if
!have_work
}
}
const SEALING_TIMEOUT_IN_BLOCKS : u64 = 5;
@ -232,29 +252,48 @@ impl MinerService for Miner {
Vec<Result<TransactionImportResult, Error>>
where T: Fn(&Address) -> AccountDetails {
let mut transaction_queue = self.transaction_queue.lock().unwrap();
transaction_queue.add_all(transactions, fetch_account)
let is_local = false;
transactions.into_iter()
.map(|tx| transaction_queue.add(tx, &fetch_account, is_local))
.collect()
}
fn import_own_transaction<T>(&self, transaction: SignedTransaction, fetch_account: T) ->
fn import_own_transaction<T>(&self, chain: &BlockChainClient, transaction: SignedTransaction, fetch_account: T) ->
Result<TransactionImportResult, Error>
where T: Fn(&Address) -> AccountDetails {
let hash = transaction.hash();
trace!(target: "own_tx", "Importing transaction: {:?}", transaction);
let mut transaction_queue = self.transaction_queue.lock().unwrap();
let import = transaction_queue.add(transaction, &fetch_account);
let imported = {
// Be sure to release the lock before we call enable_and_prepare_sealing
let mut transaction_queue = self.transaction_queue.lock().unwrap();
let import = transaction_queue.add(transaction, &fetch_account, true);
match import {
Ok(ref res) => {
trace!(target: "own_tx", "Imported transaction to {:?} (hash: {:?})", res, hash);
trace!(target: "own_tx", "Status: {:?}", transaction_queue.status());
},
Err(ref e) => {
trace!(target: "own_tx", "Failed to import transaction {:?} (hash: {:?})", e, hash);
trace!(target: "own_tx", "Status: {:?}", transaction_queue.status());
},
match import {
Ok(ref res) => {
trace!(target: "own_tx", "Imported transaction to {:?} (hash: {:?})", res, hash);
trace!(target: "own_tx", "Status: {:?}", transaction_queue.status());
},
Err(ref e) => {
trace!(target: "own_tx", "Failed to import transaction {:?} (hash: {:?})", e, hash);
trace!(target: "own_tx", "Status: {:?}", transaction_queue.status());
},
}
import
};
if imported.is_ok() {
// Make sure to do it after transaction is imported and lock is droped.
// We need to create pending block and enable sealing
let prepared = self.enable_and_prepare_sealing(chain);
// If new block has not been prepared (means we already had one)
// we need to update sealing
if !prepared {
self.update_sealing(chain);
}
}
import
imported
}
fn pending_transactions_hashes(&self) -> Vec<H256> {
@ -279,8 +318,12 @@ impl MinerService for Miner {
fn update_sealing(&self, chain: &BlockChainClient) {
if self.sealing_enabled.load(atomic::Ordering::Relaxed) {
let current_no = chain.chain_info().best_block_number;
let has_local_transactions = self.transaction_queue.lock().unwrap().has_local_pending_transactions();
let last_request = *self.sealing_block_last_request.lock().unwrap();
let should_disable_sealing = !self.force_sealing && current_no > last_request && current_no - last_request > SEALING_TIMEOUT_IN_BLOCKS;
let should_disable_sealing = !self.force_sealing
&& !has_local_transactions
&& current_no > last_request
&& current_no - last_request > SEALING_TIMEOUT_IN_BLOCKS;
if should_disable_sealing {
trace!(target: "miner", "Miner sleeping (current {}, last {})", current_no, last_request);
@ -294,19 +337,8 @@ impl MinerService for Miner {
fn map_sealing_work<F, T>(&self, chain: &BlockChainClient, f: F) -> Option<T> where F: FnOnce(&ClosedBlock) -> T {
trace!(target: "miner", "map_sealing_work: entering");
let have_work = self.sealing_work.lock().unwrap().peek_last_ref().is_some();
trace!(target: "miner", "map_sealing_work: have_work={}", have_work);
if !have_work {
self.sealing_enabled.store(true, atomic::Ordering::Relaxed);
self.prepare_sealing(chain);
}
let mut sealing_block_last_request = self.sealing_block_last_request.lock().unwrap();
let best_number = chain.chain_info().best_block_number;
if *sealing_block_last_request != best_number {
trace!(target: "miner", "map_sealing_work: Miner received request (was {}, now {}) - waking up.", *sealing_block_last_request, best_number);
*sealing_block_last_request = best_number;
}
self.enable_and_prepare_sealing(chain);
trace!(target: "miner", "map_sealing_work: sealing prepared");
let mut sealing_work = self.sealing_work.lock().unwrap();
let ret = sealing_work.use_last_ref();
trace!(target: "miner", "map_sealing_work: leaving use_last_ref={:?}", ret.as_ref().map(|b| b.block().fields().header.hash()));
@ -358,10 +390,9 @@ impl MinerService for Miner {
for tx in &txs {
let _sender = tx.sender();
}
let mut transaction_queue = self.transaction_queue.lock().unwrap();
let _ = transaction_queue.add_all(txs, |a| AccountDetails {
let _ = self.import_transactions(txs, |a| AccountDetails {
nonce: chain.nonce(a),
balance: chain.balance(a)
balance: chain.balance(a),
});
});
}

177
miner/src/transaction_queue.rs
View File

@ -53,8 +53,8 @@
//! };
//!
//! let mut txq = TransactionQueue::new();
//! txq.add(st2.clone(), &default_nonce).unwrap();
//! txq.add(st1.clone(), &default_nonce).unwrap();
//! txq.add(st2.clone(), &default_nonce, false).unwrap();
//! txq.add(st1.clone(), &default_nonce, false).unwrap();
//!
//! // Check status
//! assert_eq!(txq.status().pending, 2);
@ -106,6 +106,8 @@ struct TransactionOrder {
gas_price: U256,
/// Hash to identify associated transaction
hash: H256,
/// Set to true if transaction has been dispatched localy
is_local: bool,
}
@ -115,6 +117,7 @@ impl TransactionOrder {
nonce_height: tx.nonce() - base_nonce,
gas_price: tx.transaction.gas_price,
hash: tx.hash(),
is_local: tx.is_local,
}
}
@ -156,13 +159,17 @@ impl Ord for TransactionOrder {
/// Verified transaction (with sender)
struct VerifiedTransaction {
transaction: SignedTransaction
/// Transaction
transaction: SignedTransaction,
/// Is it local transaction?
is_local: bool,
}
impl VerifiedTransaction {
fn new(transaction: SignedTransaction) -> Result<Self, Error> {
fn new(transaction: SignedTransaction, is_local: bool) -> Result<Self, Error> {
try!(transaction.sender());
Ok(VerifiedTransaction {
transaction: transaction
transaction: transaction,
is_local: is_local,
})
}
@ -384,17 +391,8 @@ impl TransactionQueue {
}
}
/// Adds all signed transactions to queue to be verified and imported
pub fn add_all<T>(&mut self, txs: Vec<SignedTransaction>, fetch_account: T) -> Vec<Result<TransactionImportResult, Error>>
where T: Fn(&Address) -> AccountDetails {
txs.into_iter()
.map(|tx| self.add(tx, &fetch_account))
.collect()
}
/// Add signed transaction to queue to be verified and imported
pub fn add<T>(&mut self, tx: SignedTransaction, fetch_account: &T) -> Result<TransactionImportResult, Error>
pub fn add<T>(&mut self, tx: SignedTransaction, fetch_account: &T, is_local: bool) -> Result<TransactionImportResult, Error>
where T: Fn(&Address) -> AccountDetails {
trace!(target: "miner", "Importing: {:?}", tx.hash());
@ -423,7 +421,7 @@ impl TransactionQueue {
}));
}
let vtx = try!(VerifiedTransaction::new(tx));
let vtx = try!(VerifiedTransaction::new(tx, is_local));
let client_account = fetch_account(&vtx.sender());
let cost = vtx.transaction.value + vtx.transaction.gas_price * vtx.transaction.gas;
@ -556,6 +554,11 @@ impl TransactionQueue {
.collect()
}
/// Returns true if there is at least one local transaction pending
pub fn has_local_pending_transactions(&self) -> bool {
self.current.by_priority.iter().any(|tx| tx.is_local)
}
/// Finds transaction in the queue by hash (if any)
pub fn find(&self, hash: &H256) -> Option<SignedTransaction> {
match self.by_hash.get(hash) { Some(transaction_ref) => Some(transaction_ref.transaction.clone()), None => None }
@ -817,12 +820,12 @@ mod test {
limit: 1
};
let (tx1, tx2) = new_txs(U256::from(1));
let tx1 = VerifiedTransaction::new(tx1).unwrap();
let tx2 = VerifiedTransaction::new(tx2).unwrap();
let tx1 = VerifiedTransaction::new(tx1, false).unwrap();
let tx2 = VerifiedTransaction::new(tx2, false).unwrap();
let mut by_hash = {
let mut x = HashMap::new();
let tx1 = VerifiedTransaction::new(tx1.transaction.clone()).unwrap();
let tx2 = VerifiedTransaction::new(tx2.transaction.clone()).unwrap();
let tx1 = VerifiedTransaction::new(tx1.transaction.clone(), false).unwrap();
let tx2 = VerifiedTransaction::new(tx2.transaction.clone(), false).unwrap();
x.insert(tx1.hash(), tx1);
x.insert(tx2.hash(), tx2);
x
@ -858,12 +861,12 @@ mod test {
// Create two transactions with same nonce
// (same hash)
let (tx1, tx2) = new_txs(U256::from(0));
let tx1 = VerifiedTransaction::new(tx1).unwrap();
let tx2 = VerifiedTransaction::new(tx2).unwrap();
let tx1 = VerifiedTransaction::new(tx1, false).unwrap();
let tx2 = VerifiedTransaction::new(tx2, false).unwrap();
let by_hash = {
let mut x = HashMap::new();
let tx1 = VerifiedTransaction::new(tx1.transaction.clone()).unwrap();
let tx2 = VerifiedTransaction::new(tx2.transaction.clone()).unwrap();
let tx1 = VerifiedTransaction::new(tx1.transaction.clone(), false).unwrap();
let tx2 = VerifiedTransaction::new(tx2.transaction.clone(), false).unwrap();
x.insert(tx1.hash(), tx1);
x.insert(tx2.hash(), tx2);
x
@ -895,12 +898,12 @@ mod test {
!U256::zero() };
// First insert one transaction to future
let res = txq.add(tx, &prev_nonce);
let res = txq.add(tx, &prev_nonce, false);
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_nonce);
let res = txq.add(tx2.clone(), &default_nonce, false);
// and then there should be only one transaction in current (the one with higher gas_price)
assert_eq!(unwrap_tx_err(res), TransactionError::TooCheapToReplace);
@ -919,7 +922,7 @@ mod test {
let tx = new_tx();
// when
let res = txq.add(tx, &default_nonce);
let res = txq.add(tx, &default_nonce, false);
// then
assert_eq!(res.unwrap(), TransactionImportResult::Current);
@ -951,7 +954,7 @@ mod test {
txq.set_gas_limit(limit);
// when
let res = txq.add(tx, &default_nonce);
let res = txq.add(tx, &default_nonce, false);
// then
assert_eq!(unwrap_tx_err(res), TransactionError::GasLimitExceeded {
@ -975,7 +978,7 @@ mod test {
};
// when
let res = txq.add(tx, &account);
let res = txq.add(tx, &account, false);
// then
assert_eq!(unwrap_tx_err(res), TransactionError::InsufficientBalance {
@ -995,7 +998,7 @@ mod test {
txq.set_minimal_gas_price(tx.gas_price + U256::one());
// when
let res = txq.add(tx, &default_nonce);
let res = txq.add(tx, &default_nonce, false);
// then
assert_eq!(unwrap_tx_err(res), TransactionError::InsufficientGasPrice {
@ -1026,7 +1029,7 @@ mod test {
decode(s.as_raw())
};
// when
let res = txq.add(stx, &default_nonce);
let res = txq.add(stx, &default_nonce, false);
// then
assert!(res.is_err());
@ -1040,8 +1043,8 @@ mod test {
let (tx, tx2) = new_txs(U256::from(1));
// when
txq.add(tx.clone(), &default_nonce).unwrap();
txq.add(tx2.clone(), &default_nonce).unwrap();
txq.add(tx.clone(), &default_nonce, false).unwrap();
txq.add(tx2.clone(), &default_nonce, false).unwrap();
// then
let top = txq.top_transactions();
@ -1058,8 +1061,8 @@ mod test {
let (tx, tx2) = new_txs(U256::from(1));
// when
txq.add(tx.clone(), &default_nonce).unwrap();
txq.add(tx2.clone(), &default_nonce).unwrap();
txq.add(tx.clone(), &default_nonce, false).unwrap();
txq.add(tx2.clone(), &default_nonce, false).unwrap();
// then
let top = txq.pending_hashes();
@ -1076,8 +1079,8 @@ mod test {
let (tx, tx2) = new_txs(U256::from(2));
// when
let res1 = txq.add(tx.clone(), &default_nonce).unwrap();
let res2 = txq.add(tx2.clone(), &default_nonce).unwrap();
let res1 = txq.add(tx.clone(), &default_nonce, false).unwrap();
let res2 = txq.add(tx2.clone(), &default_nonce, false).unwrap();
// then
assert_eq!(res1, TransactionImportResult::Current);
@ -1100,8 +1103,8 @@ mod test {
let mut txq = TransactionQueue::new();
let (tx, tx2) = new_txs(U256::from(1));
txq.add(tx.clone(), &prev_nonce).unwrap();
txq.add(tx2.clone(), &prev_nonce).unwrap();
txq.add(tx.clone(), &prev_nonce, false).unwrap();
txq.add(tx2.clone(), &prev_nonce, false).unwrap();
assert_eq!(txq.status().future, 2);
// when
@ -1123,13 +1126,13 @@ mod test {
let tx1 = new_unsigned_tx(U256::from(124)).sign(&secret);
let tx2 = new_unsigned_tx(U256::from(125)).sign(&secret);
txq.add(tx, &default_nonce).unwrap();
txq.add(tx, &default_nonce, false).unwrap();
assert_eq!(txq.status().pending, 1);
txq.add(tx2, &default_nonce).unwrap();
txq.add(tx2, &default_nonce, false).unwrap();
assert_eq!(txq.status().future, 1);
// when
txq.add(tx1, &default_nonce).unwrap();
txq.add(tx1, &default_nonce, false).unwrap();
// then
let stats = txq.status();
@ -1142,8 +1145,8 @@ mod test {
// given
let mut txq2 = TransactionQueue::new();
let (tx, tx2) = new_txs(U256::from(3));
txq2.add(tx.clone(), &default_nonce).unwrap();
txq2.add(tx2.clone(), &default_nonce).unwrap();
txq2.add(tx.clone(), &default_nonce, false).unwrap();
txq2.add(tx2.clone(), &default_nonce, false).unwrap();
assert_eq!(txq2.status().pending, 1);
assert_eq!(txq2.status().future, 1);
@ -1164,10 +1167,10 @@ mod test {
let mut txq = TransactionQueue::new();
let (tx, tx2) = new_txs(U256::from(1));
let tx3 = new_tx();
txq.add(tx2.clone(), &default_nonce).unwrap();
txq.add(tx2.clone(), &default_nonce, false).unwrap();
assert_eq!(txq.status().future, 1);
txq.add(tx3.clone(), &default_nonce).unwrap();
txq.add(tx.clone(), &default_nonce).unwrap();
txq.add(tx3.clone(), &default_nonce, false).unwrap();
txq.add(tx.clone(), &default_nonce, false).unwrap();
assert_eq!(txq.status().pending, 3);
// when
@ -1186,8 +1189,8 @@ mod test {
let (tx, tx2) = new_txs(U256::one());
// add
txq.add(tx2.clone(), &default_nonce).unwrap();
txq.add(tx.clone(), &default_nonce).unwrap();
txq.add(tx2.clone(), &default_nonce, false).unwrap();
txq.add(tx.clone(), &default_nonce, false).unwrap();
let stats = txq.status();
assert_eq!(stats.pending, 2);
@ -1206,11 +1209,11 @@ mod test {
let (tx, tx2) = new_txs(U256::one());
let sender = tx.sender().unwrap();
let nonce = tx.nonce;
txq.add(tx.clone(), &default_nonce).unwrap();
txq.add(tx.clone(), &default_nonce, false).unwrap();
assert_eq!(txq.status().pending, 1);
// when
let res = txq.add(tx2.clone(), &default_nonce);
let res = txq.add(tx2.clone(), &default_nonce, false);
// then
let t = txq.top_transactions();
@ -1229,13 +1232,13 @@ mod test {
let (tx1, tx2) = new_txs(U256::one());
let sender = tx1.sender().unwrap();
let nonce = tx1.nonce;
txq.add(tx1.clone(), &default_nonce).unwrap();
txq.add(tx2.clone(), &default_nonce).unwrap();
txq.add(tx1.clone(), &default_nonce, false).unwrap();
txq.add(tx2.clone(), &default_nonce, false).unwrap();
assert_eq!(txq.status().pending, 2);
assert_eq!(txq.last_nonce(&sender), Some(nonce + U256::one()));
// when
let res = txq.add(tx.clone(), &default_nonce);
let res = txq.add(tx.clone(), &default_nonce, false);
// then
assert_eq!(res.unwrap(), TransactionImportResult::Current);
@ -1249,14 +1252,14 @@ mod test {
txq.current.set_limit(10);
let (tx1, tx2) = new_txs_with_gas_price_diff(U256::from(4), U256::from(1));
let (tx3, tx4) = new_txs_with_gas_price_diff(U256::from(4), U256::from(2));
txq.add(tx1.clone(), &default_nonce).unwrap();
txq.add(tx3.clone(), &default_nonce).unwrap();
txq.add(tx1.clone(), &default_nonce, false).unwrap();
txq.add(tx3.clone(), &default_nonce, false).unwrap();
assert_eq!(txq.status().pending, 2);
// when
txq.add(tx2.clone(), &default_nonce).unwrap();
txq.add(tx2.clone(), &default_nonce, false).unwrap();
assert_eq!(txq.status().future, 1);
txq.add(tx4.clone(), &default_nonce).unwrap();
txq.add(tx4.clone(), &default_nonce, false).unwrap();
// then
assert_eq!(txq.status().future, 1);
@ -1270,7 +1273,7 @@ mod test {
let fetch_last_nonce = |_a: &Address| AccountDetails{ nonce: last_nonce, balance: !U256::zero() };
// when
let res = txq.add(tx, &fetch_last_nonce);
let res = txq.add(tx, &fetch_last_nonce, false);
// then
assert_eq!(unwrap_tx_err(res), TransactionError::Old);
@ -1286,12 +1289,12 @@ mod test {
balance: !U256::zero() };
let mut txq = TransactionQueue::new();
let (_tx1, tx2) = new_txs(U256::from(1));
txq.add(tx2.clone(), &default_nonce).unwrap();
txq.add(tx2.clone(), &default_nonce, false).unwrap();
assert_eq!(txq.status().future, 1);
assert_eq!(txq.status().pending, 0);
// when
let res = txq.add(tx2.clone(), &nonce);
let res = txq.add(tx2.clone(), &nonce, false);
// then
assert_eq!(unwrap_tx_err(res), TransactionError::AlreadyImported);
@ -1305,15 +1308,15 @@ mod test {
// given
let mut txq = TransactionQueue::new();
let (tx1, tx2) = new_txs(U256::from(1));
txq.add(tx1.clone(), &default_nonce).unwrap();
txq.add(tx2.clone(), &default_nonce).unwrap();
txq.add(tx1.clone(), &default_nonce, false).unwrap();
txq.add(tx2.clone(), &default_nonce, false).unwrap();
assert_eq!(txq.status().pending, 2);
// when
txq.remove_invalid(&tx1.hash(), &default_nonce);
assert_eq!(txq.status().pending, 0);
assert_eq!(txq.status().future, 1);
txq.add(tx1.clone(), &default_nonce).unwrap();
txq.add(tx1.clone(), &default_nonce, false).unwrap();
// then
let stats = txq.status();
@ -1327,10 +1330,10 @@ mod test {
let mut txq = TransactionQueue::new();
let (tx, tx2) = new_txs(U256::from(1));
let tx3 = new_tx();
txq.add(tx2.clone(), &default_nonce).unwrap();
txq.add(tx2.clone(), &default_nonce, false).unwrap();
assert_eq!(txq.status().future, 1);
txq.add(tx3.clone(), &default_nonce).unwrap();
txq.add(tx.clone(), &default_nonce).unwrap();
txq.add(tx3.clone(), &default_nonce, false).unwrap();
txq.add(tx.clone(), &default_nonce, false).unwrap();
assert_eq!(txq.status().pending, 3);
// when
@ -1357,8 +1360,8 @@ mod test {
};
// when
txq.add(tx, &default_nonce).unwrap();
txq.add(tx2, &default_nonce).unwrap();
txq.add(tx, &default_nonce, false).unwrap();
txq.add(tx2, &default_nonce, false).unwrap();
// then
let stats = txq.status();
@ -1385,10 +1388,10 @@ mod test {
};
// when
txq.add(tx1, &default_nonce).unwrap();
txq.add(tx2, &default_nonce).unwrap();
txq.add(tx1, &default_nonce, false).unwrap();
txq.add(tx2, &default_nonce, false).unwrap();
assert_eq!(txq.status().future, 1);
txq.add(tx0, &default_nonce).unwrap();
txq.add(tx0, &default_nonce, false).unwrap();
// then
let stats = txq.status();
@ -1406,8 +1409,8 @@ mod test {
!U256::zero() };
let mut txq = TransactionQueue::new();
let (tx1, tx2) = new_txs(U256::one());
txq.add(tx1.clone(), &previous_nonce).unwrap();
txq.add(tx2, &previous_nonce).unwrap();
txq.add(tx1.clone(), &previous_nonce, false).unwrap();
txq.add(tx2, &previous_nonce, false).unwrap();
assert_eq!(txq.status().future, 2);
// when
@ -1438,7 +1441,7 @@ mod test {
let details = |_a: &Address| AccountDetails { nonce: nonce, balance: !U256::zero() };
// when
txq.add(tx, &details).unwrap();
txq.add(tx, &details, false).unwrap();
// then
assert_eq!(txq.last_nonce(&from), Some(nonce));
@ -1453,7 +1456,7 @@ mod test {
let details1 = |_a: &Address| AccountDetails { nonce: nonce1, balance: !U256::zero() };
// Insert first transaction
txq.add(tx1, &details1).unwrap();
txq.add(tx1, &details1, false).unwrap();
// when
txq.remove_all(tx2.sender().unwrap(), nonce2 + U256::one());
@ -1473,9 +1476,9 @@ mod test {
// when
// Insert first transaction
assert_eq!(txq.add(tx1, &details1).unwrap(), TransactionImportResult::Current);
assert_eq!(txq.add(tx1, &details1, false).unwrap(), TransactionImportResult::Current);
// Second should go to future
assert_eq!(txq.add(tx2, &details1).unwrap(), TransactionImportResult::Future);
assert_eq!(txq.add(tx2, &details1, false).unwrap(), TransactionImportResult::Future);
// Now block is imported
txq.remove_all(sender, nonce2 - U256::from(1));
// tx2 should be not be promoted to current
@ -1484,6 +1487,22 @@ mod test {
// then
assert_eq!(txq.last_nonce(&sender), None);
}
#[test]
fn should_return_true_if_there_is_local_transaction_pending() {
// given
let mut txq = TransactionQueue::new();
let (tx1, tx2) = new_txs(U256::from(1));
assert_eq!(txq.has_local_pending_transactions(), false);
// when
assert_eq!(txq.add(tx1, &default_nonce, false).unwrap(), TransactionImportResult::Current);
assert_eq!(txq.has_local_pending_transactions(), false);
assert_eq!(txq.add(tx2, &default_nonce, true).unwrap(), TransactionImportResult::Current);
// then
assert_eq!(txq.has_local_pending_transactions(), true);
}
}

2
rpc/src/v1/impls/eth.rs
View File

@ -173,7 +173,7 @@ impl<C, S, A, M, EM> EthClient<C, S, A, M, EM>
let client = take_weak!(self.client);
let miner = take_weak!(self.miner);
miner.import_own_transaction(signed_transaction, |a: &Address| {
miner.import_own_transaction(client.deref(), signed_transaction, |a: &Address| {
AccountDetails {
nonce: client.nonce(&a),
balance: client.balance(&a),

2
rpc/src/v1/tests/helpers/miner_service.rs
View File

@ -124,7 +124,7 @@ impl MinerService for TestMinerService {
}
/// Imports transactions to transaction queue.
fn import_own_transaction<T>(&self, transaction: SignedTransaction, _fetch_account: T) ->
fn import_own_transaction<T>(&self, _chain: &BlockChainClient, transaction: SignedTransaction, _fetch_account: T) ->
Result<TransactionImportResult, Error>
where T: Fn(&Address) -> AccountDetails {
// lets assume that all txs are valid