-
Notifications
You must be signed in to change notification settings - Fork 979
fix: rebroadcast txs recovered from reorg cache #3900
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: staging
Are you sure you want to change the base?
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -283,19 +283,31 @@ 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::<Vec<_>>(); | ||
| debug!( | ||
| "reconcile_reorg_cache: size: {}, block: {:?} ...", | ||
| 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); | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. tx_accepted() is fine for a single tx, but here it runs in a loop under the pool write lock. Each peer queue has at most 100 free slots and silently drops overflow while reporting success, so a reorg recovering more than SEND_CHANNEL_CAP txs can lose later announcements. Could we schedule these through a paced bulk-relay path outside the pool lock? |
||
| rebroadcast += 1; | ||
| } | ||
| } | ||
| debug!( | ||
| "reconcile_reorg_cache: block: {:?} ... done.", | ||
| header.hash() | ||
| "reconcile_reorg_cache: block: {:?} ... done (rebroadcast {})", | ||
| header.hash(), | ||
| rebroadcast | ||
| ); | ||
| Ok(()) | ||
| } | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -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<std::sync::Mutex<Vec<Hash>>>, | ||
| } | ||
|
|
||
| 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<B, P>( | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. this duplicates the full PoolConfig literal from init_transaction_pool for one caller. Could the existing helper delegate here, or both use a shared test config, so they cannot drift? |
||
| chain: Arc<B>, | ||
| adapter: Arc<P>, | ||
| ) -> TransactionPool<B, P> | ||
| 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<K>( | ||
| keychain: &K, | ||
| header: &BlockHeader, | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,97 @@ | ||
| // Copyright 2021 The Grin Developers | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 2026? |
||
| // | ||
| // 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(); | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. The recording adapter only counts callbacks. Could we add a real fork/reorg test for the production trigger, plus a bounded-queue adapter using SEND_CHANNEL_CAP to cover overflow? Otherwise the test misses the actual delivery failure mode. |
||
| 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()); | ||
| } | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The later discussion in #3489 explicitly questions unconditional rebroadcasting and suggests opt-in/default-off. This PR also leaves the issue’s retention-period request unchanged. Could we settle that policy here and use Refs #3489 rather than closing it unless both parts are handled?