From 0887b1c2da93a9ec23e9dd0267f76b71f43df258 Mon Sep 17 00:00:00 2001 From: iho Date: Thu, 9 Jul 2026 11:47:31 +0300 Subject: [PATCH] fix: rebroadcast txs recovered from reorg cache After a reorg we re-applied cached transactions to the local mempool but never announced them to peers, so recovered txs could remain private to the node. Call tx_accepted for each successfully re-accepted entry so the network learns about them again. Closes #3489 --- pool/src/transaction_pool.rs | 18 ++++- pool/tests/common.rs | 53 ++++++++++++++- pool/tests/reorg_cache_rebroadcast.rs | 97 +++++++++++++++++++++++++++ 3 files changed, 164 insertions(+), 4 deletions(-) create mode 100644 pool/tests/reorg_cache_rebroadcast.rs diff --git a/pool/src/transaction_pool.rs b/pool/src/transaction_pool.rs index b15252d2f2..5db737e6cf 100644 --- a/pool/src/transaction_pool.rs +++ b/pool/src/transaction_pool.rs @@ -283,6 +283,11 @@ where debug!("truncate_reorg_cache: size: {}", cache.len()); } + /// Re-apply cached transactions to the mempool after a reorg. + /// + /// Txs that are successfully re-accepted are also rebroadcast via the pool + /// adapter so peers learn about them again. Previously we only re-added to + /// the local pool and never announced them on the network (see #3489). pub fn reconcile_reorg_cache(&mut self, header: &BlockHeader) -> Result<(), PoolError> { let entries = self.reorg_cache.read().iter().cloned().collect::>(); debug!( @@ -290,12 +295,19 @@ where entries.len(), header.hash(), ); + let mut rebroadcast = 0usize; for entry in entries { - let _ = self.add_to_txpool(&entry, header); + // Only rebroadcast txs that were actually re-accepted (already-present + // or invalid-on-new-tip txs are skipped without network spam). + if self.add_to_txpool(&entry, header).is_ok() { + self.adapter.tx_accepted(&entry); + rebroadcast += 1; + } } debug!( - "reconcile_reorg_cache: block: {:?} ... done.", - header.hash() + "reconcile_reorg_cache: block: {:?} ... done (rebroadcast {})", + header.hash(), + rebroadcast ); Ok(()) } diff --git a/pool/tests/common.rs b/pool/tests/common.rs index 55a230d155..19bafe0aee 100644 --- a/pool/tests/common.rs +++ b/pool/tests/common.rs @@ -17,7 +17,7 @@ use self::chain::types::{NoopAdapter, Options}; use self::chain::Chain; use self::core::consensus; -use self::core::core::hash::Hash; +use self::core::core::hash::{Hash, Hashed}; use self::core::core::{ Block, BlockHeader, BlockSums, Inputs, KernelFeatures, OutputIdentifier, Transaction, TxKernel, }; @@ -170,6 +170,57 @@ where ) } +/// Pool adapter that records `tx_accepted` calls (used to assert rebroadcasts). +pub struct RecordingPoolAdapter { + pub accepted: Arc>>, +} + +impl RecordingPoolAdapter { + pub fn new() -> Self { + RecordingPoolAdapter { + accepted: Arc::new(std::sync::Mutex::new(Vec::new())), + } + } + + pub fn accepted_count(&self) -> usize { + self.accepted.lock().unwrap().len() + } + + pub fn clear(&self) { + self.accepted.lock().unwrap().clear(); + } +} + +impl PoolAdapter for RecordingPoolAdapter { + fn tx_accepted(&self, entry: &PoolEntry) { + self.accepted.lock().unwrap().push(entry.tx.hash()); + } + fn stem_tx_accepted(&self, _entry: &PoolEntry) -> Result<(), PoolError> { + Ok(()) + } +} + +pub fn init_transaction_pool_with_adapter( + chain: Arc, + adapter: Arc

, +) -> TransactionPool +where + B: BlockChain, + P: PoolAdapter, +{ + TransactionPool::new( + PoolConfig { + accept_fee_base: default_accept_fee_base(), + reorg_cache_period: 30, + max_pool_size: 50, + max_stempool_size: 50, + mineable_max_weight: 10_000, + }, + chain, + adapter, + ) +} + pub fn test_transaction_spending_coinbase( keychain: &K, header: &BlockHeader, diff --git a/pool/tests/reorg_cache_rebroadcast.rs b/pool/tests/reorg_cache_rebroadcast.rs new file mode 100644 index 0000000000..e23c85829a --- /dev/null +++ b/pool/tests/reorg_cache_rebroadcast.rs @@ -0,0 +1,97 @@ +// Copyright 2021 The Grin Developers +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! After a reorg, txs recovered from the reorg cache must be rebroadcast so the +//! network learns about them again (not only re-added to the local mempool). + +pub mod common; + +use self::core::global; +use self::keychain::{ExtKeychain, Keychain}; +use crate::common::*; +use grin_core as core; +use grin_keychain as keychain; +use grin_util as util; +use std::sync::Arc; + +#[test] +fn reorg_cache_rebroadcasts_reaccepted_txs() { + util::init_test_logger(); + global::set_local_chain_type(global::ChainTypes::AutomatedTesting); + global::set_local_accept_fee_base(1); + let keychain: ExtKeychain = Keychain::from_random_seed(false).unwrap(); + + let db_root = "target/.reorg_cache_rebroadcast"; + clean_output_dir(db_root.into()); + + let genesis = genesis_block(&keychain); + let chain = Arc::new(init_chain(db_root, genesis)); + + let adapter = Arc::new(RecordingPoolAdapter::new()); + let mut pool = init_transaction_pool_with_adapter( + Arc::new(ChainAdapter { + chain: chain.clone(), + }), + adapter.clone(), + ); + + // Mine past HF4 and create spendable coinbase outputs. + add_some_blocks(&chain, 4 * 3, &keychain); + let header_1 = chain.get_header_by_height(1).unwrap(); + let initial_tx = + test_transaction_spending_coinbase(&keychain, &header_1, vec![1_000, 2_000, 3_000]); + add_block(&chain, &[initial_tx], &keychain); + let header = chain.head_header().unwrap(); + + // Two independent pool txs. + let tx_a = test_transaction(&keychain, vec![1_000], vec![800]); + let tx_b = test_transaction(&keychain, vec![2_000], vec![1_500]); + + adapter.clear(); + pool.add_to_pool(test_source(), tx_a.clone(), false, &header) + .unwrap(); + pool.add_to_pool(test_source(), tx_b.clone(), false, &header) + .unwrap(); + // Initial accept calls: one per tx. + assert_eq!(adapter.accepted_count(), 2); + assert_eq!(pool.total_size(), 2); + assert_eq!(pool.reorg_cache.read().len(), 2); + + // Simulate reorg: empty the pool while leaving reorg_cache intact. + pool.txpool.entries.clear(); + assert_eq!(pool.total_size(), 0); + assert_eq!(pool.reorg_cache.read().len(), 2); + + // Reconcile reorg cache against current tip — txs should re-enter and rebroadcast. + adapter.clear(); + pool.reconcile_reorg_cache(&header).unwrap(); + assert_eq!(pool.total_size(), 2); + assert_eq!( + adapter.accepted_count(), + 2, + "each successfully re-accepted tx must be rebroadcast" + ); + + // Second reconcile: txs already in pool fail re-add; no extra rebroadcasts. + adapter.clear(); + pool.reconcile_reorg_cache(&header).unwrap(); + assert_eq!(pool.total_size(), 2); + assert_eq!( + adapter.accepted_count(), + 0, + "already-present txs must not rebroadcast" + ); + + clean_output_dir(db_root.into()); +}