Fix transaction pruning in tx-pool (#6276)

The `tree_route` generated by the import notification is only from the
old best block to the new best parent. This means, it does not contain
the new best block in `enacted()`. We need to prune the transactions of
the new best block "manually" to fix this bug.

Besides that, this pr also changed the `id` parameter of the `NewBlock`
chain event to `hash`. The hash of a block is unique in contrast to the
block number. (Block id can either be number or hash)
This commit is contained in:
Bastian Köcher
2020-06-08 12:38:19 +02:00
committed by GitHub
parent 84cdb02963
commit 663cd09be9
11 changed files with 134 additions and 81 deletions
+2
View File
@@ -6745,10 +6745,12 @@ dependencies = [
"parity-scale-codec",
"parity-util-mem",
"parking_lot 0.10.2",
"sc-block-builder",
"sc-client-api",
"sc-transaction-graph",
"sp-api",
"sp-blockchain",
"sp-consensus",
"sp-core",
"sp-keyring",
"sp-runtime",
+1 -1
View File
@@ -490,7 +490,7 @@ mod tests {
service.transaction_pool().maintain(
ChainEvent::NewBlock {
is_new_best: true,
id: parent_id.clone(),
hash: parent_header.hash(),
tree_route: None,
header: parent_header.clone(),
},
+20 -1
View File
@@ -234,7 +234,7 @@ pub struct BlockImportNotification<Block: BlockT> {
pub header: Block::Header,
/// Is this the new best block.
pub is_new_best: bool,
/// Tree route from old best to new best.
/// Tree route from old best to new best parent.
///
/// If `None`, there was no re-org while importing.
pub tree_route: Option<Arc<sp_blockchain::TreeRoute<Block>>>,
@@ -248,3 +248,22 @@ pub struct FinalityNotification<Block: BlockT> {
/// Imported block header.
pub header: Block::Header,
}
impl<B: BlockT> From<BlockImportNotification<B>> for sp_transaction_pool::ChainEvent<B> {
fn from(n: BlockImportNotification<B>) -> Self {
Self::NewBlock {
is_new_best: n.is_new_best,
hash: n.hash,
header: n.header,
tree_route: n.tree_route,
}
}
}
impl<B: BlockT> From<FinalityNotification<B>> for sp_transaction_pool::ChainEvent<B> {
fn from(n: FinalityNotification<B>) -> Self {
Self::Finalized {
hash: n.hash,
}
}
}
@@ -351,11 +351,11 @@ mod tests {
}.into_signed_tx()
}
fn chain_event<B: BlockT>(block_number: u64, header: B::Header) -> ChainEvent<B>
fn chain_event<B: BlockT>(header: B::Header) -> ChainEvent<B>
where NumberFor<B>: From<u64>
{
ChainEvent::NewBlock {
id: BlockId::Number(block_number.into()),
hash: header.hash(),
tree_route: None,
is_new_best: true,
header,
@@ -380,8 +380,9 @@ mod tests {
futures::executor::block_on(
txpool.maintain(chain_event(
0,
client.header(&BlockId::Number(0u64)).expect("header get error").expect("there should be header")
client.header(&BlockId::Number(0u64))
.expect("header get error")
.expect("there should be header")
))
);
@@ -470,7 +471,6 @@ mod tests {
futures::executor::block_on(
txpool.maintain(chain_event(
0,
client.header(&BlockId::Number(0u64))
.expect("header get error")
.expect("there should be header"),
@@ -574,8 +574,9 @@ mod tests {
futures::executor::block_on(
txpool.maintain(chain_event(
0,
client.header(&BlockId::Number(0u64)).expect("header get error").expect("there should be header")
client.header(&BlockId::Number(0u64))
.expect("header get error")
.expect("there should be header")
))
);
@@ -585,8 +586,9 @@ mod tests {
futures::executor::block_on(
txpool.maintain(chain_event(
1,
client.header(&BlockId::Number(1)).expect("header get error").expect("there should be header")
client.header(&BlockId::Number(1))
.expect("header get error")
.expect("there should be header")
))
);
@@ -413,9 +413,10 @@ mod tests {
assert!(client.header(&BlockId::Number(0)).unwrap().is_some());
assert!(pool.submit_one(&BlockId::Number(1), SOURCE, uxt(Alice, 1)).await.is_ok());
let header = client.header(&BlockId::Number(1)).expect("db error").expect("imported above");
pool.maintain(sp_transaction_pool::ChainEvent::NewBlock {
id: BlockId::Number(1),
header: client.header(&BlockId::Number(1)).expect("db error").expect("imported above"),
hash: header.hash(),
header,
is_new_best: true,
tree_route: None,
}).await;
+3 -8
View File
@@ -55,7 +55,7 @@ use std::{
};
use wasm_timer::SystemTime;
use sc_telemetry::{telemetry, SUBSTRATE_INFO};
use sp_transaction_pool::{MaintainedTransactionPool, ChainEvent};
use sp_transaction_pool::MaintainedTransactionPool;
use prometheus_endpoint::Registry;
use sc_client_db::{Backend, DatabaseSettings};
use sp_core::traits::CodeExecutor;
@@ -1042,14 +1042,9 @@ ServiceBuilder<
{
let txpool = Arc::downgrade(&transaction_pool);
let mut import_stream = client.import_notification_stream().map(|n| ChainEvent::NewBlock {
id: BlockId::Hash(n.hash),
header: n.header,
tree_route: n.tree_route,
is_new_best: n.is_new_best,
}).fuse();
let mut import_stream = client.import_notification_stream().map(Into::into).fuse();
let mut finality_stream = client.finality_notification_stream()
.map(|n| ChainEvent::Finalized::<TBl> { hash: n.hash })
.map(Into::into)
.fuse();
let events = async move {
@@ -786,7 +786,7 @@ impl<B, E, Block, RA> Client<B, E, Block, RA> where
NewBlockState::Normal
};
let tree_route = if is_new_best {
let tree_route = if is_new_best && info.best_hash != parent_hash {
let route_from_best = sp_blockchain::tree_route(
self.backend.blockchain(),
info.best_hash,
@@ -36,5 +36,7 @@ wasm-timer = "0.2"
assert_matches = "1.3.0"
hex = "0.4"
sp-keyring = { version = "2.0.0-rc2", path = "../../primitives/keyring" }
sp-consensus = { version = "0.8.0-rc2", path = "../../primitives/consensus/common" }
substrate-test-runtime-transaction-pool = { version = "2.0.0-rc2", path = "../../test-utils/runtime/transaction-pool" }
substrate-test-runtime-client = { version = "2.0.0-rc2", path = "../../test-utils/runtime/client" }
sc-block-builder = { version = "0.8.0-rc2", path = "../block-builder" }
+9 -5
View File
@@ -468,10 +468,11 @@ impl<PoolApi, Block> MaintainedTransactionPool for BasicPool<PoolApi, Block>
{
fn maintain(&self, event: ChainEvent<Self::Block>) -> Pin<Box<dyn Future<Output=()> + Send>> {
match event {
ChainEvent::NewBlock { id, tree_route, is_new_best, .. } => {
ChainEvent::NewBlock { hash, tree_route, is_new_best, .. } => {
let pool = self.pool.clone();
let api = self.api.clone();
let id = BlockId::hash(hash);
let block_number = match api.block_id_to_number(&id) {
Ok(Some(number)) => number,
_ => {
@@ -495,13 +496,13 @@ impl<PoolApi, Block> MaintainedTransactionPool for BasicPool<PoolApi, Block>
async move {
// If there is a tree route, we use this to prune known tx based on the enacted
// blocks and otherwise we only prune known txs if the block is
// the new best block.
// blocks.
if let Some(ref tree_route) = tree_route {
future::join_all(
tree_route
.enacted()
.iter().map(|h|
.iter()
.map(|h|
prune_known_txs_for_block(
BlockId::Hash(h.hash.clone()),
&*api,
@@ -509,7 +510,10 @@ impl<PoolApi, Block> MaintainedTransactionPool for BasicPool<PoolApi, Block>
),
),
).await;
} else if is_new_best {
}
// If this is a new best block, we need to prune its transactions from the pool.
if is_new_best {
prune_known_txs_for_block(id.clone(), &*api, &*pool).await;
}
@@ -18,7 +18,7 @@
use crate::*;
use sp_transaction_pool::TransactionStatus;
use futures::executor::block_on;
use futures::executor::{block_on, block_on_stream};
use txpool::{self, Pool};
use sp_runtime::{
generic::BlockId,
@@ -26,11 +26,15 @@ use sp_runtime::{
};
use substrate_test_runtime_client::{
runtime::{Block, Hash, Index, Header, Extrinsic, Transfer}, AccountKeyring::*,
ClientBlockImportExt,
};
use substrate_test_runtime_transaction_pool::{TestApi, uxt};
use futures::{prelude::*, task::Poll};
use codec::Encode;
use std::collections::BTreeSet;
use sc_client_api::client::BlockchainEvents;
use sc_block_builder::BlockBuilderProvider;
use sp_consensus::BlockOrigin;
fn pool() -> Pool<TestApi> {
Pool::new(Default::default(), TestApi::with_alice_nonce(209).into())
@@ -50,16 +54,6 @@ fn maintained_pool() -> (
(pool, thread_pool, notifier)
}
fn header(number: u64) -> Header {
Header {
number,
digest: Default::default(),
extrinsics_root: Default::default(),
parent_hash: Default::default(),
state_root: Default::default(),
}
}
const SOURCE: TransactionSource = TransactionSource::External;
#[test]
@@ -153,7 +147,7 @@ fn only_prune_on_new_best() {
assert_eq!(pool.status().ready, 1);
let event = ChainEvent::NewBlock {
id: BlockId::Number(1),
hash: header.hash(),
is_new_best: false,
header: header.clone(),
tree_route: None,
@@ -163,7 +157,7 @@ fn only_prune_on_new_best() {
let header = pool.api.push_block(2, vec![uxt]);
let event = ChainEvent::NewBlock {
id: BlockId::Number(2),
hash: header.hash(),
is_new_best: true,
header: header.clone(),
tree_route: None,
@@ -209,12 +203,12 @@ fn should_correctly_prune_transactions_providing_more_than_one_tag() {
assert_eq!(pool.validated_pool().status().future, 2);
}
fn block_event(id: u64) -> ChainEvent<Block> {
fn block_event(header: Header) -> ChainEvent<Block> {
ChainEvent::NewBlock {
id: BlockId::number(id),
hash: header.hash(),
is_new_best: true,
tree_route: None,
header: header(id),
header,
}
}
@@ -223,10 +217,10 @@ fn block_event_with_retracted(
retracted_start: Hash,
api: &TestApi,
) -> ChainEvent<Block> {
let tree_route = api.tree_route(retracted_start, header.hash()).expect("Tree route exists");
let tree_route = api.tree_route(retracted_start, header.parent_hash).expect("Tree route exists");
ChainEvent::NewBlock {
id: BlockId::hash(header.hash()),
hash: header.hash(),
is_new_best: true,
tree_route: Some(Arc::new(tree_route)),
header,
@@ -242,9 +236,9 @@ fn should_prune_old_during_maintenance() {
block_on(pool.submit_one(&BlockId::number(0), SOURCE, xt.clone())).expect("1. Imported");
assert_eq!(pool.status().ready, 1);
pool.api.push_block(1, vec![xt.clone()]);
let header = pool.api.push_block(1, vec![xt.clone()]);
block_on(pool.maintain(block_event(1)));
block_on(pool.maintain(block_event(header)));
assert_eq!(pool.status().ready, 0);
}
@@ -259,9 +253,9 @@ fn should_revalidate_during_maintenance() {
assert_eq!(pool.status().ready, 2);
assert_eq!(pool.api.validation_requests().len(), 2);
pool.api.push_block(1, vec![xt1.clone()]);
let header = pool.api.push_block(1, vec![xt1.clone()]);
block_on(pool.maintain(block_event(1)));
block_on(pool.maintain(block_event(header)));
assert_eq!(pool.status().ready, 1);
block_on(notifier.next());
@@ -317,17 +311,17 @@ fn should_revalidate_transaction_multiple_times() {
block_on(pool.submit_one(&BlockId::number(0), SOURCE, xt.clone())).expect("1. Imported");
assert_eq!(pool.status().ready, 1);
pool.api.push_block(1, vec![xt.clone()]);
let header = pool.api.push_block(1, vec![xt.clone()]);
block_on(pool.maintain(block_event(1)));
block_on(pool.maintain(block_event(header)));
block_on(pool.submit_one(&BlockId::number(0), SOURCE, xt.clone())).expect("1. Imported");
assert_eq!(pool.status().ready, 1);
pool.api.push_block(2, vec![]);
let header = pool.api.push_block(2, vec![]);
pool.api.add_invalid(&xt);
block_on(pool.maintain(block_event(2)));
block_on(pool.maintain(block_event(header)));
block_on(notifier.next());
assert_eq!(pool.status().ready, 0);
@@ -345,15 +339,15 @@ fn should_revalidate_across_many_blocks() {
block_on(pool.submit_one(&BlockId::number(1), SOURCE, xt2.clone())).expect("1. Imported");
assert_eq!(pool.status().ready, 2);
pool.api.push_block(1, vec![]);
block_on(pool.maintain(block_event(1)));
let header = pool.api.push_block(1, vec![]);
block_on(pool.maintain(block_event(header)));
block_on(notifier.next());
block_on(pool.submit_one(&BlockId::number(2), SOURCE, xt3.clone())).expect("1. Imported");
assert_eq!(pool.status().ready, 3);
pool.api.push_block(2, vec![xt1.clone()]);
block_on(pool.maintain(block_event(2)));
let header = pool.api.push_block(2, vec![xt1.clone()]);
block_on(pool.maintain(block_event(header)));
block_on(notifier.next());
assert_eq!(pool.status().ready, 2);
@@ -398,7 +392,8 @@ fn should_push_watchers_during_maintaince() {
pool.api.add_invalid(&tx4);
// clear timer events if any
block_on(pool.maintain(block_event(0)));
let header = pool.api.push_block(1, vec![]);
block_on(pool.maintain(block_event(header)));
block_on(notifier.next());
// then
@@ -415,8 +410,9 @@ fn should_push_watchers_during_maintaince() {
);
// when
let header_hash = pool.api.push_block(1, vec![tx0, tx1, tx2]).hash();
block_on(pool.maintain(block_event(1)));
let header = pool.api.push_block(2, vec![tx0, tx1, tx2]);
let header_hash = header.hash();
block_on(pool.maintain(block_event(header)));
let event = ChainEvent::Finalized { hash: header_hash.clone() };
block_on(pool.maintain(event));
@@ -472,7 +468,7 @@ fn finalization() {
let header = pool.api.chain().read().block_by_number.get(&2).unwrap()[0].header().clone();
let event = ChainEvent::NewBlock {
id: BlockId::Hash(header.hash()),
hash: header.hash(),
is_new_best: true,
header: header.clone(),
tree_route: None,
@@ -524,7 +520,7 @@ fn fork_aware_finalization() {
assert_eq!(pool.status().ready, 1);
let event = ChainEvent::NewBlock {
id: BlockId::Number(2),
hash: header.hash(),
is_new_best: true,
header: header.clone(),
tree_route: None,
@@ -544,7 +540,7 @@ fn fork_aware_finalization() {
).expect("1. Imported");
assert_eq!(pool.status().ready, 1);
let event = ChainEvent::NewBlock {
id: BlockId::Hash(header.hash()),
hash: header.hash(),
is_new_best: true,
header: header.clone(),
tree_route: None,
@@ -563,7 +559,7 @@ fn fork_aware_finalization() {
let header = pool.api.push_block_with_parent(c2, vec![from_bob.clone()]);
let event = ChainEvent::NewBlock {
id: BlockId::Hash(header.hash()),
hash: header.hash(),
is_new_best: true,
header: header.clone(),
tree_route: None,
@@ -601,7 +597,7 @@ fn fork_aware_finalization() {
canon_watchers.push((w, header.hash()));
let event = ChainEvent::NewBlock {
id: BlockId::Hash(header.hash()),
hash: header.hash(),
is_new_best: true,
header: header.clone(),
tree_route: None,
@@ -620,7 +616,7 @@ fn fork_aware_finalization() {
let header = pool.api.push_block(5, vec![from_dave, from_bob]);
e1 = header.hash();
let event = ChainEvent::NewBlock {
id: BlockId::Hash(header.hash()),
hash: header.hash(),
is_new_best: true,
header: header.clone(),
tree_route: None,
@@ -702,7 +698,7 @@ fn resubmit_tx_of_fork_that_is_not_part_of_retracted() {
assert_eq!(pool.status().ready, 1);
let event = ChainEvent::NewBlock {
id: BlockId::Number(2),
hash: header.hash(),
is_new_best: true,
header: header.clone(),
tree_route: None,
@@ -720,7 +716,7 @@ fn resubmit_tx_of_fork_that_is_not_part_of_retracted() {
let header = pool.api.push_block(2, vec![tx1.clone()]);
assert_eq!(pool.status().ready, 1);
let event = ChainEvent::NewBlock {
id: BlockId::Hash(header.hash()),
hash: header.hash(),
is_new_best: false,
header: header.clone(),
tree_route: None,
@@ -773,7 +769,7 @@ fn resubmit_from_retracted_fork() {
assert_eq!(pool.status().ready, 1);
let event = ChainEvent::NewBlock {
id: BlockId::Number(2),
hash: header.hash(),
is_new_best: true,
header: header.clone(),
tree_route: None,
@@ -789,7 +785,7 @@ fn resubmit_from_retracted_fork() {
).expect("1. Imported");
let header = pool.api.push_block(3, vec![tx1.clone()]);
let event = ChainEvent::NewBlock {
id: BlockId::Hash(header.hash()),
hash: header.hash(),
is_new_best: true,
header: header.clone(),
tree_route: None,
@@ -805,7 +801,7 @@ fn resubmit_from_retracted_fork() {
).expect("1. Imported");
let header = pool.api.push_block(4, vec![tx2.clone()]);
let event = ChainEvent::NewBlock {
id: BlockId::Hash(header.hash()),
hash: header.hash(),
is_new_best: true,
header: header.clone(),
tree_route: None,
@@ -822,7 +818,7 @@ fn resubmit_from_retracted_fork() {
).expect("1. Imported");
let header = pool.api.push_block(2, vec![tx3.clone()]);
let event = ChainEvent::NewBlock {
id: BlockId::Hash(header.hash()),
hash: header.hash(),
is_new_best: false,
header: header.clone(),
tree_route: None,
@@ -839,7 +835,7 @@ fn resubmit_from_retracted_fork() {
).expect("1. Imported");
let header = pool.api.push_block_with_parent(d1.clone(), vec![tx4.clone()]);
let event = ChainEvent::NewBlock {
id: BlockId::Hash(header.hash()),
hash: header.hash(),
is_new_best: false,
header: header.clone(),
tree_route: None,
@@ -886,12 +882,12 @@ fn ready_set_should_not_resolve_before_block_update() {
#[test]
fn ready_set_should_resolve_after_block_update() {
let (pool, _guard, _notifier) = maintained_pool();
pool.api.push_block(1, vec![]);
let header = pool.api.push_block(1, vec![]);
let xt1 = uxt(Alice, 209);
block_on(pool.submit_one(&BlockId::number(1), SOURCE, xt1.clone())).expect("1. Imported");
block_on(pool.maintain(block_event(1)));
block_on(pool.maintain(block_event(header)));
assert!(pool.ready_at(1).now_or_never().is_some());
}
@@ -899,7 +895,7 @@ fn ready_set_should_resolve_after_block_update() {
#[test]
fn ready_set_should_eventually_resolve_when_block_update_arrives() {
let (pool, _guard, _notifier) = maintained_pool();
pool.api.push_block(1, vec![]);
let header = pool.api.push_block(1, vec![]);
let xt1 = uxt(Alice, 209);
@@ -913,7 +909,7 @@ fn ready_set_should_eventually_resolve_when_block_update_arrives() {
panic!("Ready set should not be ready before block update!");
}
block_on(pool.maintain(block_event(1)));
block_on(pool.maintain(block_event(header)));
match ready_set_future.poll_unpin(&mut context) {
Poll::Pending => {
@@ -949,7 +945,11 @@ fn should_not_accept_old_signatures() {
"c427eb672e8c441c86d31f1a81b22b43102058e9ce237cabe9897ea5099ffd426cd1c6a1f4f2869c3df57901d36bedcb295657adb3a4355add86ed234eb83108"
).expect("hex invalid")[..]).expect("signature construction failed");
let xt = Extrinsic::Transfer { transfer, signature: old_singature, exhaust_resources_when_not_first: false };
let xt = Extrinsic::Transfer {
transfer,
signature: old_singature,
exhaust_resources_when_not_first: false,
};
assert_matches::assert_matches!(
block_on(pool.submit_one(&BlockId::number(0), SOURCE, xt.clone())),
@@ -959,3 +959,31 @@ fn should_not_accept_old_signatures() {
"Should be invalid transaction with bad proof",
);
}
#[test]
fn import_notification_to_pool_maintain_works() {
let mut client = Arc::new(substrate_test_runtime_client::new());
let pool = Arc::new(
BasicPool::new_test(Arc::new(FullChainApi::new(client.clone()))).0
);
// Prepare the extrisic, push it to the pool and check that it was added.
let xt = uxt(Alice, 0);
block_on(pool.submit_one(&BlockId::number(0), SOURCE, xt.clone())).expect("1. Imported");
assert_eq!(pool.status().ready, 1);
let mut import_stream = block_on_stream(client.import_notification_stream());
// Build the block with the transaction included
let mut block_builder = client.new_block(Default::default()).unwrap();
block_builder.push(xt).unwrap();
let block = block_builder.build().unwrap().block;
client.import(BlockOrigin::Own, block).unwrap();
// Get the notification of the block import and maintain the pool with it,
// Now, the pool should not contain any transactions.
let evt = import_stream.next().expect("Importing a block leads to an event");
block_on(pool.maintain(evt.into()));
assert_eq!(pool.status().ready, 0);
}
@@ -251,11 +251,11 @@ pub enum ChainEvent<B: BlockT> {
NewBlock {
/// Is this the new best block.
is_new_best: bool,
/// Id of the just imported block.
id: BlockId<B>,
/// Hash of the block.
hash: B::Hash,
/// Header of the just imported block
header: B::Header,
/// Tree route from old best to new best that was calculated on import.
/// Tree route from old best to new best parent that was calculated on import.
///
/// If `None`, no re-org happened on import.
tree_route: Option<Arc<sp_blockchain::TreeRoute<B>>>,