From 6b85ddcd17d2342000592f59656eb2ddc12cb4cb Mon Sep 17 00:00:00 2001 From: Marek Date: Wed, 2 Sep 2026 13:05:43 +0200 Subject: [PATCH] perf(rpc): rebuild the block template when the mempool changes The block template updater task rebuilt every five seconds whether or not anything had changed, so a new transaction waited up to that long to reach miners while an idle chain paid for rebuilds nobody needed. Subscribe to the mempool change channel instead: additions rebuild, and a burst of them coalesces into one rebuild. A rebuild reads the whole mempool, re-runs ZIP-317 selection, and rebuilds the coinbase, so changes that cannot alter the template do not trigger one. The `Invalidated` kind also fires for transactions that failed verification and were never in the mempool, so it only rebuilds when the template's long poll ID covers the transaction. That ID is derived from every ID in the mempool rather than the transactions ZIP-317 selected, so the cache stores the whole set. Overflowing the change channel compares the mempool with that set rather than rebuilding, or a peer sending invalid transactions fast enough to make the channel lag would buy the rebuilds the filter denies it. A thirty second backstop keeps `cur_time` current, which Testnet's minimum-difficulty rule depends on, and bounds a lost notification. --- .../zebra-rpc-Added-20260901-141500.yaml | 5 +- .../zebra-rpc-Changed-20260901-141501.yaml | 5 +- .../zebrad-Changed-20260901-141502.yaml | 6 +- zebra-rpc/src/methods.rs | 11 +- zebra-rpc/src/methods/tests/vectors.rs | 646 +++++++++++++++++- .../types/get_block_template/precompute.rs | 236 ++++++- .../get_block_template/precompute/tests.rs | 266 +++++++- zebrad/src/commands/start.rs | 2 +- 8 files changed, 1118 insertions(+), 59 deletions(-) diff --git a/.changes/unreleased/zebra-rpc-Added-20260901-141500.yaml b/.changes/unreleased/zebra-rpc-Added-20260901-141500.yaml index 67ea18814c0..149a1900cf1 100644 --- a/.changes/unreleased/zebra-rpc-Added-20260901-141500.yaml +++ b/.changes/unreleased/zebra-rpc-Added-20260901-141500.yaml @@ -1,6 +1,7 @@ project: zebra-rpc kind: Added -body: 'Method `RpcImpl::spawn_block_template_updater()`, which spawns a task that keeps a block - template precomputed for the current chain tip, and returns `None` when mining isn''t configured +body: 'Method `RpcImpl::spawn_block_template_updater()`, which takes a + `MempoolTxSubscriber` and spawns a task that keeps a block template precomputed for the current + chain tip, and returns `None` when mining isn''t configured ([#11370](https://github.com/ZcashFoundation/zebra/issues/11370)).' time: 2026-09-01T14:15:00.000000000Z diff --git a/.changes/unreleased/zebra-rpc-Changed-20260901-141501.yaml b/.changes/unreleased/zebra-rpc-Changed-20260901-141501.yaml index 549371265eb..9486c2ac8cd 100644 --- a/.changes/unreleased/zebra-rpc-Changed-20260901-141501.yaml +++ b/.changes/unreleased/zebra-rpc-Changed-20260901-141501.yaml @@ -4,7 +4,8 @@ body: 'The `getblocktemplate` RPC answers from the template precomputed by `RpcImpl::spawn_block_template_updater()`, so a call validates the chain tip against the state instead of reading the mempool, selecting transactions, and building a coinbase transaction. A long polling call waits on that template, and returns when the task publishes a new one, when the - chain tip changes, or when `max_time` is reached. A precomputed template''s transactions can be a - few seconds behind the mempool, but it always extends the tip the state has committed + chain tip changes, or when `max_time` is reached. The task rebuilds when the chain tip changes and + when the mempool changes, rather than on a fixed interval, so a precomputed template''s + transactions trail the mempool by a rebuild, and it always extends the tip the state has committed ([#11370](https://github.com/ZcashFoundation/zebra/issues/11370)).' time: 2026-09-01T14:15:01.000000000Z diff --git a/.changes/unreleased/zebrad-Changed-20260901-141502.yaml b/.changes/unreleased/zebrad-Changed-20260901-141502.yaml index a199af0a2ca..590c58e1408 100644 --- a/.changes/unreleased/zebrad-Changed-20260901-141502.yaml +++ b/.changes/unreleased/zebrad-Changed-20260901-141502.yaml @@ -2,8 +2,8 @@ project: zebrad kind: Changed body: 'The `getblocktemplate` RPC returns a block template that Zebra keeps ready for the current chain tip, rather than assembling one while the miner waits. Zebra keeps one ready whenever a - miner address is configured and either the RPC server or the internal miner is enabled, refreshing - it on every chain tip change and every few seconds, so a template can be a few seconds behind the - mempool but is never behind the chain + miner address is configured and either the RPC server or the internal miner is enabled, rebuilding + it when the chain tip changes and when the mempool changes, instead of every few seconds, so a new + transaction reaches miners in the next rebuild and a template is never behind the chain ([#11370](https://github.com/ZcashFoundation/zebra/issues/11370)).' time: 2026-09-01T14:15:02.000000000Z diff --git a/zebra-rpc/src/methods.rs b/zebra-rpc/src/methods.rs index 73f4096a3ad..6e8bffaa919 100644 --- a/zebra-rpc/src/methods.rs +++ b/zebra-rpc/src/methods.rs @@ -85,7 +85,7 @@ use zebra_consensus::{ funding_stream_address, router::service_trait::BlockVerifierService, RouterError, }; use zebra_network::{address_book_peers::AddressBookPeers, types::PeerServices, PeerSocketAddr}; -use zebra_node_services::mempool::{self, CreatedOrSpent, MempoolService}; +use zebra_node_services::mempool::{self, CreatedOrSpent, MempoolService, MempoolTxSubscriber}; use zebra_state::{ AnyTx, HashOrHeight, OutputLocation, ReadRequest, ReadResponse, ReadState as ReadStateService, State as StateService, TransactionLocation, @@ -1008,8 +1008,14 @@ where /// `getblocktemplate` calls don't have to read the state and the mempool, select transactions, /// and build a coinbase transaction. /// + /// The task rebuilds when the chain tip changes and when `mempool_change` reports a change + /// that affects the template, so a new transaction reaches miners without waiting for a timer. + /// /// Returns `None` if mining isn't configured. - pub fn spawn_block_template_updater(&self) -> Option> { + pub fn spawn_block_template_updater( + &self, + mempool_change: MempoolTxSubscriber, + ) -> Option> { let miner_params = self.gbt.miner_params()?.clone(); let template_cache = self.gbt.template_cache()?.clone(); @@ -1020,6 +1026,7 @@ where self.gbt.coinbase_cache(), template_cache, self.mempool.clone(), + mempool_change.subscribe(), self.read_state.clone(), self.latest_chain_tip.clone(), self.gbt.sync_status(), diff --git a/zebra-rpc/src/methods/tests/vectors.rs b/zebra-rpc/src/methods/tests/vectors.rs index a243de986d8..d9b7efea77d 100644 --- a/zebra-rpc/src/methods/tests/vectors.rs +++ b/zebra-rpc/src/methods/tests/vectors.rs @@ -9,6 +9,7 @@ use std::{ }; use futures::FutureExt; +use tokio::sync::broadcast; use tower::buffer::Buffer; use zcash_address::{ToAddress, ZcashAddress}; @@ -36,7 +37,10 @@ use zebra_consensus::MAX_BLOCK_SIGOPS; use zebra_network::{ address_book_peers::MockAddressBookPeers, types::PeerServices, PeerSocketAddr, }; -use zebra_node_services::BoxError; +use zebra_node_services::{ + mempool::{MempoolChange, MempoolTxSubscriber}, + BoxError, +}; use zebra_state::{ GetBlockTemplateChainInfo, IntoDisk, LatestChainTip, ReadRequest, ReadResponse, ReadStateService, @@ -2880,8 +2884,11 @@ async fn getblocktemplate_precomputed() { let (read_state_responder, mempool_responder) = spawn_responders(tip_height, tip_hash); + // Holding the sender keeps the updater's change receiver open. + let (mempool_change_tx, _mempool_change_rx) = broadcast::channel(100); + let updater = rpc - .spawn_block_template_updater() + .spawn_block_template_updater(MempoolTxSubscriber::new(mempool_change_tx.clone())) .expect("mining is configured"); // Wait for the updater task to precompute a template for the mock chain tip. @@ -3107,8 +3114,11 @@ async fn getblocktemplate_long_poll_waits_for_a_new_template() { } }); + // Holding the sender keeps the updater's change receiver open. + let (mempool_change_tx, _mempool_change_rx) = broadcast::channel(100); + let updater = rpc - .spawn_block_template_updater() + .spawn_block_template_updater(MempoolTxSubscriber::new(mempool_change_tx.clone())) .expect("mining is configured"); let template_cache = rpc @@ -3301,8 +3311,11 @@ async fn getblocktemplate_ignores_precomputed_template_when_tip_channel_lags_sta // Fill the cache with a template for the tip both the state and the channel agree on. let (read_state_responder, mempool_responder) = spawn_responders(tip_height, tip_hash); + // Holding the sender keeps the updater's change receiver open. + let (mempool_change_tx, _mempool_change_rx) = broadcast::channel(100); + let updater = rpc - .spawn_block_template_updater() + .spawn_block_template_updater(MempoolTxSubscriber::new(mempool_change_tx.clone())) .expect("mining is configured"); let template_cache = rpc @@ -3342,6 +3355,631 @@ async fn getblocktemplate_ignores_precomputed_template_when_tip_channel_lags_sta updater.abort(); } +/// How long the tests below wait for a rebuild they expect. +/// +/// [`wait_for_count`] returns as soon as the rebuild lands, so this is only a ceiling. It has to +/// stay well under `BACKSTOP_REFRESH`, or the updater's own backstop rebuild could satisfy an +/// assertion that the change under test was supposed to satisfy, and the test would pass against +/// an updater that ignored the change entirely. +const REBUILD_TIMEOUT: Duration = Duration::from_secs(5); + +/// Waits for `counter` to reach `target`, polling until `timeout` elapses. +/// +/// Returns `true` as soon as it does, so a test that expects an event pays only for the event. +/// Returns `false` if it never does, which is how the tests below assert that a change the updater +/// should ignore cost no rebuild: a rebuild it decided to make starts one [`MEMPOOL_DEBOUNCE`] +/// after the change that prompted it, so a window longer than that catches one. +async fn wait_for_count(counter: &AtomicUsize, target: usize, timeout: Duration) -> bool { + tokio::time::timeout(timeout, async { + while counter.load(Ordering::SeqCst) < target { + tokio::time::sleep(Duration::from_millis(5)).await; + } + }) + .await + .is_ok() +} + +/// Checks that a mempool change reaches the block template updater and makes it rebuild. +/// +/// Which changes are worth a rebuild, how a burst is coalesced, and what a lagging change channel +/// costs are covered by the wait-phase tests in `precompute::tests`, which read the decision +/// straight off a paused clock. This test covers the wiring those tests can't: a change sent on +/// the mempool's channel reaching a running updater task, and that task reading the mempool again. +#[tokio::test(flavor = "multi_thread")] +async fn getblocktemplate_precompute_rebuilds_on_a_mempool_change() { + let _init_guard = zebra_test::init(); + + let net = Network::Mainnet; + + let request_delay = Duration::from_secs(60); + let mempool: MockService<_, _, _, BoxError> = MockService::build() + .with_max_request_delay(request_delay) + .for_unit_tests(); + let state: MockService<_, _, _, BoxError> = MockService::build().for_unit_tests(); + let read_state: MockService<_, _, _, BoxError> = MockService::build() + .with_max_request_delay(request_delay) + .for_unit_tests(); + + let mut mock_sync_status = MockSyncStatus::default(); + mock_sync_status.set_is_close_to_tip(true); + + let mining_conf = mining::Config { + miner_address: Some(ZcashAddress::from_transparent_p2pkh( + NetworkType::from(NetworkKind::from(&net)), + [0x7e; 20], + )), + extra_coinbase_data: None, + miner_memo: None, + internal_miner: false, + }; + + let tip_height = NetworkUpgrade::Nu5 + .activation_height(&net) + .expect("nu5 activation height"); + let tip_hash = + Hash::from_hex("0000000000d723156d9b65ffcf4984da7a19675ed7e2f06d9e5d5188af087bf8").unwrap(); + + let (mock_tip, mock_tip_sender) = MockChainTip::new(); + mock_tip_sender.send_best_tip_height(tip_height); + mock_tip_sender.send_best_tip_hash(tip_hash); + mock_tip_sender.send_estimated_distance_to_network_chain_tip(Some(0)); + + let (_tx, rx) = tokio::sync::watch::channel(None); + let (rpc, _) = RpcImpl::new( + net.clone(), + mining_conf, + Default::default(), + "0.0.1", + "RPC test", + Buffer::new(mempool.clone(), 1), + state.clone(), + Buffer::new(read_state.clone(), 1), + MockService::build().for_unit_tests(), + mock_sync_status, + mock_tip, + MockAddressBookPeers::default(), + rx, + None, + ); + + // Counts the mempool reads a rebuild makes, which is one `FullTransactions` request each. + let rebuilds = Arc::new(AtomicUsize::new(0)); + + let chain_history_root = fake_history_tree(&net).hash(); + let read_state_responder = tokio::spawn({ + let mut read_state = read_state.clone(); + async move { + loop { + read_state + .expect_request_that(|req| { + matches!(req, ReadRequest::ChainInfo | ReadRequest::Tip) + }) + .await + .respond_with(move |req| match req { + ReadRequest::ChainInfo => { + ReadResponse::ChainInfo(GetBlockTemplateChainInfo { + expected_difficulty: CompactDifficulty::from( + ExpandedDifficulty::from(U256::one()), + ), + tip_height, + tip_hash, + cur_time: DateTime32::from(1654008617), + min_time: DateTime32::from(1654008606), + max_time: DateTime32::from(1654008719), + chain_history_root, + }) + } + ReadRequest::Tip => ReadResponse::Tip(Some((tip_height, tip_hash))), + other => panic!("unexpected read state request: {other:?}"), + }); + } + } + }); + + let mempool_responder = tokio::spawn({ + let mut mempool = mempool.clone(); + let rebuilds = rebuilds.clone(); + async move { + loop { + mempool + .expect_request(mempool::Request::FullTransactions) + .await + .respond(mempool::Response::FullTransactions { + transactions: vec![], + transaction_dependencies: Default::default(), + last_seen_tip_hash: tip_hash, + }); + rebuilds.fetch_add(1, Ordering::SeqCst); + } + } + }); + + let (mempool_change_tx, _mempool_change_rx) = broadcast::channel(100); + + let updater = rpc + .spawn_block_template_updater(MempoolTxSubscriber::new(mempool_change_tx.clone())) + .expect("mining is configured"); + + let template_cache = rpc + .gbt + .template_cache() + .expect("the miner params were not overridden") + .clone(); + + tokio::time::timeout(Duration::from_secs(30), async { + while template_cache.wait_for_tip(tip_hash).await.is_none() {} + }) + .await + .expect("the updater task should precompute a template for the chain tip"); + + // The mempool responder counts a read after it answers, so wait for the first rebuild's read + // rather than assuming it has landed. + assert!( + wait_for_count(&rebuilds, 1, REBUILD_TIMEOUT).await, + "the updater should read the mempool once while precomputing the first template", + ); + let before_change = rebuilds.load(Ordering::SeqCst); + + mempool_change_tx + .send(MempoolChange::added( + [UnminedTxId::from_legacy_id(transaction::Hash([1; 32]))] + .into_iter() + .collect(), + )) + .expect("the updater task holds a receiver"); + + assert!( + wait_for_count(&rebuilds, before_change + 1, REBUILD_TIMEOUT).await, + "a mempool addition should make the updater read the mempool and rebuild", + ); + + read_state_responder.abort(); + mempool_responder.abort(); + updater.abort(); +} + +/// Checks that a long polling `getblocktemplate` call returns as soon as the block template +/// updater publishes a template for a changed mempool, rather than waiting for the RPC's own +/// mempool poll. +#[tokio::test(flavor = "multi_thread")] +async fn getblocktemplate_long_poll_returns_on_mempool_change() { + let _init_guard = zebra_test::init(); + + let net = Network::Mainnet; + + let request_delay = Duration::from_secs(60); + let mempool: MockService<_, _, _, BoxError> = MockService::build() + .with_max_request_delay(request_delay) + .for_unit_tests(); + let state: MockService<_, _, _, BoxError> = MockService::build().for_unit_tests(); + let read_state: MockService<_, _, _, BoxError> = MockService::build() + .with_max_request_delay(request_delay) + .for_unit_tests(); + + let mut mock_sync_status = MockSyncStatus::default(); + mock_sync_status.set_is_close_to_tip(true); + + let mining_conf = mining::Config { + miner_address: Some(ZcashAddress::from_transparent_p2pkh( + NetworkType::from(NetworkKind::from(&net)), + [0x7e; 20], + )), + extra_coinbase_data: None, + miner_memo: None, + internal_miner: false, + }; + + let tip_height = NetworkUpgrade::Nu5 + .activation_height(&net) + .expect("nu5 activation height"); + let tip_hash = + Hash::from_hex("0000000000d723156d9b65ffcf4984da7a19675ed7e2f06d9e5d5188af087bf8").unwrap(); + + let (mock_tip, mock_tip_sender) = MockChainTip::new(); + mock_tip_sender.send_best_tip_height(tip_height); + mock_tip_sender.send_best_tip_hash(tip_hash); + mock_tip_sender.send_estimated_distance_to_network_chain_tip(Some(0)); + + let (_tx, rx) = tokio::sync::watch::channel(None); + let (rpc, _) = RpcImpl::new( + net.clone(), + mining_conf, + Default::default(), + "0.0.1", + "RPC test", + Buffer::new(mempool.clone(), 1), + state.clone(), + Buffer::new(read_state.clone(), 1), + MockService::build().for_unit_tests(), + mock_sync_status, + mock_tip, + MockAddressBookPeers::default(), + rx, + None, + ); + + // A transaction the mempool only starts returning once the test asks it to, standing in for a + // transaction that arrives while a miner is long polling. + let tx = Arc::new(Transaction::test_v1( + vec![], + vec![], + transaction::LockTime::unlocked(), + )); + let unmined_tx = UnminedTx { + transaction: tx.clone(), + id: tx.unmined_id(), + size: tx.zcash_serialized_size(), + conventional_fee: 0.try_into().unwrap(), + }; + let new_tx_id = unmined_tx.id; + let new_tx = VerifiedUnminedTx { + conventional_actions: zip317::conventional_actions(&unmined_tx.transaction), + transaction: unmined_tx, + miner_fee: 0.try_into().unwrap(), + legacy_sigop_count: 0, + p2sh_sigop_count: 0, + unpaid_actions: 0, + fee_weight_ratio: 1.0, + time: None, + height: None, + spent_outputs: Arc::new(vec![]), + }; + + // Switched from an empty mempool to one holding `new_tx`. + let mempool_has_tx = Arc::new(AtomicBool::new(false)); + + // Counts the state tip reads, so the test can tell when the long polling call is waiting. + // + // Long polling is served from the precomputed cache, which validates the template against the + // state tip on every wake and never reads the mempool, so this is the observable that says the + // call has parked. + let tip_reads = Arc::new(AtomicUsize::new(0)); + + let chain_history_root = fake_history_tree(&net).hash(); + let read_state_responder = tokio::spawn({ + let mut read_state = read_state.clone(); + let tip_reads = tip_reads.clone(); + async move { + loop { + let tip_reads = tip_reads.clone(); + read_state + .expect_request_that(|req| { + matches!(req, ReadRequest::ChainInfo | ReadRequest::Tip) + }) + .await + .respond_with(move |req| match req { + ReadRequest::ChainInfo => { + ReadResponse::ChainInfo(GetBlockTemplateChainInfo { + expected_difficulty: CompactDifficulty::from( + ExpandedDifficulty::from(U256::one()), + ), + tip_height, + tip_hash, + cur_time: DateTime32::from(1654008617), + min_time: DateTime32::from(1654008606), + max_time: DateTime32::from(1654008719), + chain_history_root, + }) + } + ReadRequest::Tip => { + tip_reads.fetch_add(1, Ordering::SeqCst); + ReadResponse::Tip(Some((tip_height, tip_hash))) + } + other => panic!("unexpected read state request: {other:?}"), + }); + } + } + }); + + let mempool_responder = tokio::spawn({ + let mut mempool = mempool.clone(); + let mempool_has_tx = mempool_has_tx.clone(); + async move { + loop { + // Read the switch when the request is answered, not when this iteration starts: + // the responder parks in `expect_request`, so deciding earlier answers a request + // that arrives after the switch flips with the contents from before it. + let mempool_has_tx = mempool_has_tx.clone(); + let new_tx = new_tx.clone(); + + mempool + .expect_request(mempool::Request::FullTransactions) + .await + .respond_with(move |_| mempool::Response::FullTransactions { + transactions: if mempool_has_tx.load(Ordering::SeqCst) { + vec![new_tx] + } else { + vec![] + }, + transaction_dependencies: Default::default(), + last_seen_tip_hash: tip_hash, + }); + } + } + }); + + let (mempool_change_tx, _mempool_change_rx) = broadcast::channel(100); + + let updater = rpc + .spawn_block_template_updater(MempoolTxSubscriber::new(mempool_change_tx.clone())) + .expect("mining is configured"); + + let template_cache = rpc + .gbt + .template_cache() + .expect("the miner params were not overridden") + .clone(); + + tokio::time::timeout(Duration::from_secs(30), async { + while template_cache.wait_for_tip(tip_hash).await.is_none() {} + }) + .await + .expect("the updater task should precompute a template for the chain tip"); + + let template = rpc + .get_block_template(None) + .await + .expect("getblocktemplate should succeed") + .try_into_template() + .expect("getblocktemplate without parameters should return a template"); + + assert!( + template.transactions.is_empty(), + "the mempool is empty before the change, so its template should be too", + ); + + // Long poll on the template the client already has, which can only return once something + // changes. + let long_poll = tokio::spawn({ + let rpc = rpc.clone(); + let long_poll_id = template.long_poll_id; + async move { + rpc.get_block_template(Some(GetBlockTemplateParameters { + mode: GetBlockTemplateRequestMode::Template, + data: None, + capabilities: vec![], + long_poll_id: Some(long_poll_id), + _work_id: None, + })) + .await + } + }); + + // The call above validated the cached template against the state tip, and the long polling + // call does the same before it parks, so a second tip read means it is waiting. + assert!( + wait_for_count(&tip_reads, 2, REBUILD_TIMEOUT).await, + "the long polling call should validate the tip before it waits", + ); + + mempool_has_tx.store(true, Ordering::SeqCst); + mempool_change_tx + .send(MempoolChange::added([new_tx_id].into_iter().collect())) + .expect("the updater task holds a receiver"); + + // The updater debounces for half a second and then publishes, so this budget is several times + // the event-driven path, and less than the RPC's own mempool poll interval. + assert!( + Duration::from_secs(MEMPOOL_LONG_POLL_INTERVAL) > Duration::from_millis(2500), + "this test's budget has to be shorter than the poll it proves the RPC doesn't wait for", + ); + + let template = tokio::time::timeout(Duration::from_millis(2500), long_poll) + .await + .expect( + "long polling should return once the updater publishes a template, without \ + waiting for the RPC's own mempool poll", + ) + .expect("the long polling task should not panic") + .expect("getblocktemplate should succeed") + .try_into_template() + .expect("getblocktemplate in template mode should return a template"); + + assert_eq!( + template.transactions.len(), + 1, + "the template should contain the transaction that was added to the mempool", + ); + + read_state_responder.abort(); + mempool_responder.abort(); + updater.abort(); +} + +/// Checks that the block template updater rebuilds when a transaction ZIP-317 left out of the +/// template is invalidated. +/// +/// The template's long poll ID covers every transaction in the mempool, not just the selected +/// ones, so a template built from a mempool that no longer exists has to be replaced. +#[tokio::test(flavor = "multi_thread")] +async fn getblocktemplate_precompute_rebuilds_when_an_unselected_transaction_is_invalidated() { + let _init_guard = zebra_test::init(); + + let net = Network::Mainnet; + + let request_delay = Duration::from_secs(60); + let mempool: MockService<_, _, _, BoxError> = MockService::build() + .with_max_request_delay(request_delay) + .for_unit_tests(); + let state: MockService<_, _, _, BoxError> = MockService::build().for_unit_tests(); + let read_state: MockService<_, _, _, BoxError> = MockService::build() + .with_max_request_delay(request_delay) + .for_unit_tests(); + + let mut mock_sync_status = MockSyncStatus::default(); + mock_sync_status.set_is_close_to_tip(true); + + let mining_conf = mining::Config { + miner_address: Some(ZcashAddress::from_transparent_p2pkh( + NetworkType::from(NetworkKind::from(&net)), + [0x7e; 20], + )), + extra_coinbase_data: None, + miner_memo: None, + internal_miner: false, + }; + + let tip_height = NetworkUpgrade::Nu5 + .activation_height(&net) + .expect("nu5 activation height"); + let tip_hash = + Hash::from_hex("0000000000d723156d9b65ffcf4984da7a19675ed7e2f06d9e5d5188af087bf8").unwrap(); + + let (mock_tip, mock_tip_sender) = MockChainTip::new(); + mock_tip_sender.send_best_tip_height(tip_height); + mock_tip_sender.send_best_tip_hash(tip_hash); + mock_tip_sender.send_estimated_distance_to_network_chain_tip(Some(0)); + + let (_tx, rx) = tokio::sync::watch::channel(None); + let (rpc, _) = RpcImpl::new( + net.clone(), + mining_conf, + Default::default(), + "0.0.1", + "RPC test", + Buffer::new(mempool.clone(), 1), + state.clone(), + Buffer::new(read_state.clone(), 1), + MockService::build().for_unit_tests(), + mock_sync_status, + mock_tip, + MockAddressBookPeers::default(), + rx, + None, + ); + + // A mempool transaction with more sigops than a block can hold, so ZIP-317 selection can never + // put it in a template, while it still counts towards the template's long poll ID. + let tx = Arc::new(Transaction::test_v1( + vec![], + vec![], + transaction::LockTime::unlocked(), + )); + let unmined_tx = UnminedTx { + transaction: tx.clone(), + id: tx.unmined_id(), + size: tx.zcash_serialized_size(), + conventional_fee: 0.try_into().unwrap(), + }; + let unselectable_tx_id = unmined_tx.id; + let unselectable_tx = VerifiedUnminedTx { + conventional_actions: zip317::conventional_actions(&unmined_tx.transaction), + transaction: unmined_tx, + miner_fee: 0.try_into().unwrap(), + legacy_sigop_count: MAX_BLOCK_SIGOPS + 1, + p2sh_sigop_count: 0, + unpaid_actions: 0, + fee_weight_ratio: 1.0, + time: None, + height: None, + spent_outputs: Arc::new(vec![]), + }; + + let rebuilds = Arc::new(AtomicUsize::new(0)); + + let chain_history_root = fake_history_tree(&net).hash(); + let read_state_responder = tokio::spawn({ + let mut read_state = read_state.clone(); + async move { + loop { + read_state + .expect_request_that(|req| { + matches!(req, ReadRequest::ChainInfo | ReadRequest::Tip) + }) + .await + .respond_with(move |req| match req { + ReadRequest::ChainInfo => { + ReadResponse::ChainInfo(GetBlockTemplateChainInfo { + expected_difficulty: CompactDifficulty::from( + ExpandedDifficulty::from(U256::one()), + ), + tip_height, + tip_hash, + cur_time: DateTime32::from(1654008617), + min_time: DateTime32::from(1654008606), + max_time: DateTime32::from(1654008719), + chain_history_root, + }) + } + ReadRequest::Tip => ReadResponse::Tip(Some((tip_height, tip_hash))), + other => panic!("unexpected read state request: {other:?}"), + }); + } + } + }); + + let mempool_responder = tokio::spawn({ + let mut mempool = mempool.clone(); + let rebuilds = rebuilds.clone(); + async move { + loop { + mempool + .expect_request(mempool::Request::FullTransactions) + .await + .respond(mempool::Response::FullTransactions { + transactions: vec![unselectable_tx.clone()], + transaction_dependencies: Default::default(), + last_seen_tip_hash: tip_hash, + }); + rebuilds.fetch_add(1, Ordering::SeqCst); + } + } + }); + + let (mempool_change_tx, _mempool_change_rx) = broadcast::channel(100); + + let updater = rpc + .spawn_block_template_updater(MempoolTxSubscriber::new(mempool_change_tx.clone())) + .expect("mining is configured"); + + let template_cache = rpc + .gbt + .template_cache() + .expect("the miner params were not overridden") + .clone(); + + tokio::time::timeout(Duration::from_secs(30), async { + while template_cache.wait_for_tip(tip_hash).await.is_none() {} + }) + .await + .expect("the updater task should precompute a template for the chain tip"); + + let template = tokio::time::timeout(Duration::from_secs(10), rpc.get_block_template(None)) + .await + .expect("getblocktemplate should answer from the precomputed template") + .expect("getblocktemplate should succeed") + .try_into_template() + .expect("getblocktemplate without parameters should return a template"); + + assert!( + template.transactions.is_empty(), + "the mempool transaction should be too large for the block sigop limit, so this test's \ + premise requires the template to leave it out", + ); + + assert!( + wait_for_count(&rebuilds, 1, REBUILD_TIMEOUT).await, + "the updater should read the mempool once while precomputing the first template", + ); + let before_invalidation = rebuilds.load(Ordering::SeqCst); + + mempool_change_tx + .send(MempoolChange::invalidated( + [unselectable_tx_id].into_iter().collect(), + )) + .expect("the updater task holds a receiver"); + + assert!( + wait_for_count(&rebuilds, before_invalidation + 1, REBUILD_TIMEOUT).await, + "invalidating a transaction the template's long poll ID covers should rebuild the \ + template, even though ZIP-317 didn't select it", + ); + + read_state_responder.abort(); + mempool_responder.abort(); + updater.abort(); +} + #[tokio::test(flavor = "multi_thread")] async fn rpc_submitblock_errors() { let _init_guard = zebra_test::init(); diff --git a/zebra-rpc/src/methods/types/get_block_template/precompute.rs b/zebra-rpc/src/methods/types/get_block_template/precompute.rs index dc4c51da33e..4d3b5749cd0 100644 --- a/zebra-rpc/src/methods/types/get_block_template/precompute.rs +++ b/zebra-rpc/src/methods/types/get_block_template/precompute.rs @@ -11,23 +11,25 @@ //! the chain: the RPC ignores a template whose previous block hash isn't the current tip, and //! [`run()`] publishes a coinbase-only template for a new tip as soon as it sees one. -use std::{sync::Arc, time::Duration}; +use std::{collections::HashSet, sync::Arc, time::Duration}; use jsonrpsee::core::RpcResult; use tokio::{ - sync::watch, + sync::{broadcast, watch}, task::JoinHandle, time::{sleep, timeout}, }; +use tower::ServiceExt; use zebra_chain::{ amount::{Amount, NegativeOrZero}, block::{self, Height}, chain_sync_status::ChainSyncStatus, chain_tip::ChainTip, parameters::Network, + transaction::UnminedTxId, }; -use zebra_node_services::mempool::MempoolService; +use zebra_node_services::mempool::{self, MempoolChange, MempoolChangeKind, MempoolService}; use zebra_state::ReadState; use crate::{ @@ -36,9 +38,8 @@ use crate::{ }; use super::{ - check_synced_to_tip, constants::MEMPOOL_LONG_POLL_INTERVAL, fetch_chain_info, - fetch_mempool_transactions, zip317::select_mempool_transactions, BlockTemplateResponse, - CoinbaseCache, MinerParams, + check_synced_to_tip, fetch_chain_info, fetch_mempool_transactions, + zip317::select_mempool_transactions, BlockTemplateResponse, CoinbaseCache, MinerParams, }; #[cfg(test)] @@ -55,10 +56,34 @@ const NEW_TIP_TIMEOUT: Duration = Duration::from_secs(1); /// and the mempool disagree about the tip. const RETRY_DELAY: Duration = Duration::from_secs(1); +/// How long [`run()`] waits after a mempool change before rebuilding, so a burst of changes costs +/// one rebuild rather than one per transaction. +/// +/// Rebuilding reads the whole mempool, re-runs ZIP-317 selection, and rebuilds the coinbase, so it +/// is far more expensive than the change notification that triggers it. +const MEMPOOL_DEBOUNCE: Duration = Duration::from_millis(500); + +/// How long [`run()`] waits before rebuilding when nothing has changed. +/// +/// Mempool and chain tip changes drive rebuilds, so this only has to keep `cur_time` from ageing: +/// on Testnet the difficulty depends on it through the minimum-difficulty rule. It also bounds how +/// long a lost notification can stall the template. +const BACKSTOP_REFRESH: Duration = Duration::from_secs(30); + +/// A precomputed template, with the mempool it was built from. +/// +/// The template's long poll ID is derived from every ID in the mempool, not just the transactions +/// ZIP-317 selected, so deciding whether a mempool change invalidates the template needs the whole +/// set. +struct Precomputed { + template: Arc, + mempool_tx_ids: Arc>, +} + /// A block template for the block after the current chain tip, shared between [`run()`] and the /// `getblocktemplate` RPC. #[derive(Clone)] -pub(crate) struct TemplateCache(Arc>>>); +pub(crate) struct TemplateCache(Arc>>); impl Default for TemplateCache { fn default() -> Self { @@ -67,7 +92,7 @@ impl Default for TemplateCache { } /// A subscription to the templates [`run()`] publishes. -pub(crate) struct TemplateChanges(watch::Receiver>>); +pub(crate) struct TemplateChanges(watch::Receiver>); impl TemplateChanges { /// Waits for a template published since this subscription was created, or since the last wait @@ -90,7 +115,7 @@ impl TemplateCache { self.0 .borrow() .as_ref() - .is_some_and(|template| template.previous_block_hash == tip_hash) + .is_some_and(|precomputed| precomputed.template.previous_block_hash == tip_hash) } /// Returns `true` if no [`run()`] task has published a template yet. @@ -108,9 +133,33 @@ impl TemplateCache { TemplateChanges(self.0.subscribe()) } - /// Publishes `template` as the precomputed template. - fn publish(&self, template: BlockTemplateResponse) { - self.0.send_replace(Some(Arc::new(template))); + /// Returns the mempool IDs the precomputed template was built from, if there is one. + fn mempool_tx_ids(&self) -> Option>> { + self.0 + .borrow() + .as_ref() + .map(|precomputed| precomputed.mempool_tx_ids.clone()) + } + + /// Returns `true` if any of `tx_ids` was in the mempool the template was built from. + /// + /// This is the set behind the template's long poll ID, so it includes transactions ZIP-317 left + /// out: removing one of those still has to produce a new long poll ID, or a long polling miner + /// waits on work whose mempool no longer exists. + fn built_from_any(&self, tx_ids: &HashSet) -> bool { + let Some(mempool_tx_ids) = self.mempool_tx_ids() else { + return false; + }; + + tx_ids.iter().any(|tx_id| mempool_tx_ids.contains(tx_id)) + } + + /// Publishes `template`, built from the mempool holding `mempool_tx_ids`. + fn publish(&self, template: BlockTemplateResponse, mempool_tx_ids: HashSet) { + self.0.send_replace(Some(Precomputed { + template: Arc::new(template), + mempool_tx_ids: Arc::new(mempool_tx_ids), + })); } /// Returns the precomputed template if it extends `tip_hash`, waiting up to @@ -127,7 +176,7 @@ impl TemplateCache { // An empty cache means no `run()` task has published a template, so there's nothing to wait // for. - let mut template = receiver.borrow_and_update().clone()?; + let mut template = receiver.borrow_and_update().as_ref()?.template.clone(); timeout(NEW_TIP_TIMEOUT, async move { loop { @@ -136,7 +185,7 @@ impl TemplateCache { } receiver.changed().await.ok()?; - template = receiver.borrow_and_update().clone()?; + template = receiver.borrow_and_update().as_ref()?.template.clone(); } }) .await @@ -148,9 +197,9 @@ impl TemplateCache { /// Keeps `cache` filled with a block template for the current chain tip. /// /// Publishes a coinbase-only template as soon as the chain tip changes, then replaces it with a -/// template that contains mempool transactions. Refreshes that template every -/// [`MEMPOOL_LONG_POLL_INTERVAL`] seconds, so it picks up new mempool transactions and a recent -/// `cur_time`. +/// template that contains mempool transactions. Rebuilds when the chain tip changes, when the +/// mempool changes in a way that affects the template (debounced by [`MEMPOOL_DEBOUNCE`]), and +/// every [`BACKSTOP_REFRESH`] otherwise, which keeps `cur_time` current. /// /// Runs until the task is aborted. #[allow(clippy::too_many_arguments)] @@ -160,6 +209,7 @@ pub(crate) async fn run( coinbase_cache: CoinbaseCache, cache: TemplateCache, mempool: Mempool, + mut mempool_changes: broadcast::Receiver, read_state: ReadStateService, mut latest_chain_tip: Tip, sync_status: SyncStatus, @@ -212,7 +262,7 @@ pub(crate) async fn run( ) .await { - Ok(Some(template)) => cache.publish(template), + Ok(Some((template, mempool_tx_ids))) => cache.publish(template, mempool_tx_ids), // A coinbase-only template doesn't read the mempool, so it can't be out of // sync with the state. Ok(None) => {} @@ -249,13 +299,13 @@ pub(crate) async fn run( }; match built { - Ok(Some(template)) => { + Ok(Some((template, mempool_tx_ids))) => { if was_failing { tracing::info!("block template builds recovered"); was_failing = false; } - cache.publish(template) + cache.publish(template, mempool_tx_ids) } // The state and the mempool disagreed about the tip, so retry with fresh data. Ok(None) => { @@ -293,22 +343,128 @@ pub(crate) async fn run( start_precomputing_coinbase(&mut next_coinbase, &network, &miner_params, height); } - // Refresh the template when the chain tip changes, or when the mempool has had time to - // change. Miners can keep working on an old set of transactions, so they don't need to know - // about new mempool transactions immediately. - let mut tip_change = latest_chain_tip.clone(); + // Wait for something that changes the template. + if !wait_for_change(&latest_chain_tip, &mut mempool_changes, &cache, &mempool).await { + // The chain tip channel or the mempool change channel closed: Zebra is shutting down. + return; + } + } +} + +/// Waits until the precomputed template is worth rebuilding: the chain tip changed, the mempool +/// changed in a way that affects the template, or [`BACKSTOP_REFRESH`] elapsed. +/// +/// Returns `false` when a channel closes, which means Zebra is shutting down. +async fn wait_for_change( + latest_chain_tip: &Tip, + mempool_changes: &mut broadcast::Receiver, + cache: &TemplateCache, + mempool: &Mempool, +) -> bool +where + Mempool: MempoolService, + Tip: ChainTip + Clone + Send + Sync + 'static, +{ + let mut tip_change = latest_chain_tip.clone(); + let backstop = sleep(BACKSTOP_REFRESH); + tokio::pin!(backstop); + + loop { tokio::select! { biased; + tip_changed = tip_change.best_tip_changed() => { - if tip_changed.is_err() { - return; + return tip_changed.is_ok(); + } + + change = mempool_changes.recv() => { + match change { + Ok(change) if affects_template(&change, cache) => { + // Collapse the rest of the burst into this rebuild: a busy mempool + // notifies far more often than a template is worth rebuilding. + sleep(MEMPOOL_DEBOUNCE).await; + while mempool_changes.try_recv().is_ok() {} + return true; + } + // A change that can't affect the template. Keep waiting, so a peer spraying + // rejected transactions can't make us rebuild. + Ok(_) => continue, + // Changes were dropped, so we don't know what they were. + // + // # Security + // + // Rebuilding unconditionally here would undo the filter above: a peer sending + // invalid transactions fast enough overflows this channel, and every overflow + // would buy the rebuild its rejected transactions could not. So compare the + // mempool with the set the template was built from instead. + Err(broadcast::error::RecvError::Lagged(dropped)) => { + tracing::debug!(?dropped, "mempool change channel lagged"); + while mempool_changes.try_recv().is_ok() {} + + match current_mempool_tx_ids(mempool.clone()).await { + Some(current) + if Some(¤t) != cache.mempool_tx_ids().as_deref() => + { + sleep(MEMPOOL_DEBOUNCE).await; + while mempool_changes.try_recv().is_ok() {} + return true; + } + // The same mempool, or the mempool didn't answer: nothing to do, and + // the backstop still bounds how long the template can go unrefreshed. + _ => continue, + } + } + Err(broadcast::error::RecvError::Closed) => return false, } } - _ = sleep(Duration::from_secs(MEMPOOL_LONG_POLL_INTERVAL)) => {} + + _ = &mut backstop => return true, } } } +/// Returns the IDs currently in the mempool, or `None` if it didn't answer. +/// +/// `TransactionIds` is much cheaper than the `FullTransactions` a rebuild needs: it copies IDs +/// rather than whole transactions. +async fn current_mempool_tx_ids(mempool: Mempool) -> Option> +where + Mempool: MempoolService, +{ + let response = mempool + .oneshot(mempool::Request::TransactionIds) + .await + .ok()?; + + match response { + mempool::Response::TransactionIds(tx_ids) => Some(tx_ids), + _ => None, + } +} + +/// Returns `true` if `change` can change what the next template should contain. +fn affects_template(change: &MempoolChange, cache: &TemplateCache) -> bool { + match change.kind() { + // New transactions are candidates for the next template. + MempoolChangeKind::Added => true, + + // A transaction leaving the mempool only matters if the template names it: a template that + // still contains it would produce a block that can't be mined. + // + // # Security + // + // This variant also fires for transactions that failed verification and were never in the + // mempool, so rebuilding for every one of them would let a peer force sustained rebuilds + // by sending invalid transactions. Debouncing alone doesn't fix that, since the peer can + // simply keep sending. + MempoolChangeKind::Invalidated => cache.built_from_any(change.tx_ids()), + + // Mined transactions arrive with the chain tip change that mined them, which rebuilds the + // template anyway. + MempoolChangeKind::Mined => false, + } +} + /// Builds a block template for the block after the current chain tip. /// /// Selects mempool transactions if `mempool` is `Some`, and builds a coinbase-only template @@ -319,7 +475,7 @@ async fn build( coinbase_cache: &CoinbaseCache, read_state: ReadStateService, mempool: Option, -) -> RpcResult> +) -> RpcResult)>> where Mempool: MempoolService, ReadStateService: ReadState, @@ -339,11 +495,14 @@ where None => Default::default(), }; + let mempool_tx_ids: HashSet = + mempool_txs.iter().map(|tx| tx.transaction.id).collect(); + let long_poll_id = LongPollInput::new( chain_info.tip_height, chain_info.tip_hash, chain_info.max_time, - mempool_txs.iter().map(|tx| tx.transaction.id), + mempool_tx_ids.iter().copied(), ) .generate_id(); @@ -364,14 +523,17 @@ where ); // `submit_old` depends on the long poll ID the client sent, so the RPC sets it. - Some(BlockTemplateResponse::new_internal( - &network, - &coinbase_cache, - &miner_params, - &chain_info, - long_poll_id, - mempool_txs, - None, + Some(( + BlockTemplateResponse::new_internal( + &network, + &coinbase_cache, + &miner_params, + &chain_info, + long_poll_id, + mempool_txs, + None, + ), + mempool_tx_ids, )) }) .await diff --git a/zebra-rpc/src/methods/types/get_block_template/precompute/tests.rs b/zebra-rpc/src/methods/types/get_block_template/precompute/tests.rs index 66118435038..02763f822c7 100644 --- a/zebra-rpc/src/methods/types/get_block_template/precompute/tests.rs +++ b/zebra-rpc/src/methods/types/get_block_template/precompute/tests.rs @@ -1,16 +1,28 @@ -//! Fixed test vectors for the precomputed block template cache. +//! Fixed test vectors for the block template updater's wait phase. +//! +//! [`wait_for_change`] decides whether a mempool change is worth rebuilding the template for, and +//! debounces the ones that are. Every decision shows up as *when* it returns: after +//! [`MEMPOOL_DEBOUNCE`] if a change woke it, or after [`BACKSTOP_REFRESH`] if nothing did. These +//! tests run on a paused clock and assert on that duration, so each costs no real time and says +//! which of the two happened, instead of waiting to see whether a rebuild arrives. -use std::time::Duration; +use std::{collections::HashSet, time::Duration}; + +use tokio::{sync::broadcast, time::Instant}; use zcash_keys::address::Address; use zebra_chain::{ block, + chain_tip::mock::MockChainTip, parameters::{Network, NetworkUpgrade}, serialization::DateTime32, + transaction::{self, UnminedTxId}, work::difficulty::{CompactDifficulty, ExpandedDifficulty, U256}, }; +use zebra_node_services::{mempool, BoxError}; use zebra_state::GetBlockTemplateChainInfo; +use zebra_test::mock_service::{MockService, PanicAssertion}; use crate::{ config::mining::{default_miner_address, MinerAddressType}, @@ -19,9 +31,30 @@ use crate::{ use super::*; -/// Returns a template to publish. Its contents don't matter here, only that publishing it is a -/// change. -fn template() -> BlockTemplateResponse { +/// A [`MockService`] standing in for the mempool. +type MockMempool = MockService; + +/// Returns a mempool mock whose request deadline outlasts [`BACKSTOP_REFRESH`]. +/// +/// On a paused clock an unanswered request would otherwise become the earliest deadline, and the +/// mock would panic before the wait reached the backstop. +fn mock_mempool() -> MockMempool { + MockService::build() + .with_max_request_delay(BACKSTOP_REFRESH * 2) + .for_unit_tests() +} + +/// Returns a transaction ID derived from `byte`, so tests can name IDs without building +/// transactions. +fn tx_id(byte: u8) -> UnminedTxId { + UnminedTxId::from_legacy_id(transaction::Hash([byte; 32])) +} + +/// Returns a template whose long poll ID covers `mempool_tx_ids`. +/// +/// The template itself contains no transactions, so every ID here is one the template's long poll +/// ID covers without being in the template: the difference one of the tests below covers. +fn template(mempool_tx_ids: &HashSet) -> BlockTemplateResponse { let net = Network::Mainnet; let tip_height = NetworkUpgrade::Nu5 .activation_height(&net) @@ -49,7 +82,7 @@ fn template() -> BlockTemplateResponse { chain_info.tip_height, chain_info.tip_hash, chain_info.max_time, - std::iter::empty(), + mempool_tx_ids.iter().copied(), ) .generate_id(); @@ -64,6 +97,223 @@ fn template() -> BlockTemplateResponse { ) } +/// Returns a cache holding a template built from `mempool_tx_ids`. +fn cache_built_from(mempool_tx_ids: HashSet) -> TemplateCache { + let cache = TemplateCache::default(); + cache.publish(template(&mempool_tx_ids), mempool_tx_ids); + + cache +} + +/// Runs one wait phase and returns whether it asks for a rebuild, and how long it waited. +/// +/// The clock is paused, so the duration is exactly the deadline the wait ended on, which separates +/// a change waking the updater from the backstop expiring. +async fn wait_once( + cache: &TemplateCache, + mempool_changes: &mut broadcast::Receiver, + mempool: &MockMempool, +) -> (bool, Duration) { + // A tip that never changes, so only the mempool paths are under test. + let (tip, _tip_sender) = MockChainTip::new(); + let started = Instant::now(); + let rebuild = wait_for_change(&tip, mempool_changes, cache, mempool).await; + + (rebuild, started.elapsed()) +} + +/// Checks that a burst of additions costs one wait, and that the wait consumes the whole burst. +#[tokio::test(flavor = "current_thread", start_paused = true)] +async fn a_burst_of_additions_is_one_debounced_wait() { + let _init_guard = zebra_test::init(); + + let (change_sender, mut change_receiver) = broadcast::channel(200); + let mempool = mock_mempool(); + let cache = cache_built_from(HashSet::new()); + + for byte in 0..20 { + change_sender + .send(MempoolChange::added([tx_id(byte)].into_iter().collect())) + .expect("the receiver below is open"); + } + + let (rebuild, waited) = wait_once(&cache, &mut change_receiver, &mempool).await; + + assert!(rebuild, "an addition should ask for a rebuild"); + assert_eq!( + waited, MEMPOOL_DEBOUNCE, + "the wait should end one debounce after the first addition, neither immediately nor at \ + the backstop", + ); + assert!( + change_receiver.try_recv().is_err(), + "the debounce should consume the rest of the burst, so the next wait doesn't rebuild \ + again for changes this one already covered", + ); +} + +/// Checks that invalidating transactions the template was not built from doesn't wake the wait. +/// +/// The mempool sends `Invalidated` for transactions that failed verification and were never in the +/// mempool, so rebuilding for those would let a peer spend Zebra's CPU by sending invalid +/// transactions. +#[tokio::test(flavor = "current_thread", start_paused = true)] +async fn invalidating_unused_transactions_waits_for_the_backstop() { + let _init_guard = zebra_test::init(); + + let (change_sender, mut change_receiver) = broadcast::channel(200); + let mempool = mock_mempool(); + let cache = cache_built_from([tx_id(1)].into_iter().collect()); + + for byte in 100..120 { + change_sender + .send(MempoolChange::invalidated( + [tx_id(byte)].into_iter().collect(), + )) + .expect("the receiver below is open"); + } + + let (rebuild, waited) = wait_once(&cache, &mut change_receiver, &mempool).await; + + assert!(rebuild, "the backstop should ask for a rebuild"); + assert_eq!( + waited, BACKSTOP_REFRESH, + "invalidating transactions the template wasn't built from should leave the wait to the \ + backstop", + ); +} + +/// Checks that invalidating a transaction the template's long poll ID covers wakes the wait, even +/// though ZIP-317 left it out of the template. +#[tokio::test(flavor = "current_thread", start_paused = true)] +async fn invalidating_an_unselected_transaction_ends_the_wait() { + let _init_guard = zebra_test::init(); + + let (change_sender, mut change_receiver) = broadcast::channel(200); + let mempool = mock_mempool(); + + // The template holds no transactions, so this ID is in the mempool the template was built + // from without being in the template. + let unselected = tx_id(1); + let cache = cache_built_from([unselected].into_iter().collect()); + + change_sender + .send(MempoolChange::invalidated( + [unselected].into_iter().collect(), + )) + .expect("the receiver below is open"); + + let (rebuild, waited) = wait_once(&cache, &mut change_receiver, &mempool).await; + + assert!( + rebuild, + "the invalidated transaction should ask for a rebuild" + ); + assert_eq!( + waited, MEMPOOL_DEBOUNCE, + "a transaction the template's long poll ID covers should end the wait even when ZIP-317 \ + didn't select it, because the mempool the template was built from no longer exists", + ); +} + +/// Checks that overflowing the change channel doesn't wake the wait while the mempool still holds +/// what the template was built from. +/// +/// Waking on a lagged channel would undo the filter the other tests cover: a peer sending invalid +/// transactions fast enough makes the channel lag, and every overflow would buy the rebuild its +/// rejected transactions could not. +#[tokio::test(flavor = "current_thread", start_paused = true)] +async fn a_lagging_channel_with_an_unchanged_mempool_waits_for_the_backstop() { + let _init_guard = zebra_test::init(); + + let capacity = 4; + let (change_sender, mut change_receiver) = broadcast::channel(capacity); + let template_tx_ids: HashSet = [tx_id(1)].into_iter().collect(); + let cache = cache_built_from(template_tx_ids.clone()); + + // The wait and the responder have to share one mock: a `MockService` clone only sees requests + // made after it was cloned, and answering on an unrelated instance would leave the request + // unanswered, which takes the same branch as an unchanged mempool and proves nothing. + let mut responding_mempool = mock_mempool(); + let mempool = responding_mempool.clone(); + + for byte in 0..(capacity as u8 + 2) { + change_sender + .send(MempoolChange::invalidated( + [tx_id(byte)].into_iter().collect(), + )) + .expect("the receiver below is open"); + } + + // The same mempool the template was built from, so there is nothing to rebuild for. + let responder = tokio::spawn(async move { + responding_mempool + .expect_request(mempool::Request::TransactionIds) + .await + .respond(mempool::Response::TransactionIds(template_tx_ids)); + }); + + let (rebuild, waited) = wait_once(&cache, &mut change_receiver, &mempool).await; + + // Joining the responder is what keeps this test honest: it panics if the lagged channel never + // made the wait ask the mempool what it holds. + responder + .await + .expect("the lagged wait should request the mempool's transaction IDs"); + + assert!(rebuild, "the backstop should ask for a rebuild"); + assert_eq!( + waited, BACKSTOP_REFRESH, + "a lagging channel should leave the wait to the backstop while the mempool still holds \ + what the template was built from", + ); +} + +/// Checks that overflowing the change channel does wake the wait when the mempool no longer holds +/// what the template was built from, which is the case the comparison above exists to allow. +#[tokio::test(flavor = "current_thread", start_paused = true)] +async fn a_lagging_channel_with_a_changed_mempool_ends_the_wait() { + let _init_guard = zebra_test::init(); + + let capacity = 4; + let (change_sender, mut change_receiver) = broadcast::channel(capacity); + let cache = cache_built_from([tx_id(1)].into_iter().collect()); + + let mut responding_mempool = mock_mempool(); + let mempool = responding_mempool.clone(); + + for byte in 0..(capacity as u8 + 2) { + change_sender + .send(MempoolChange::invalidated( + [tx_id(byte)].into_iter().collect(), + )) + .expect("the receiver below is open"); + } + + // A different mempool from the one the template was built from. + let responder = tokio::spawn(async move { + responding_mempool + .expect_request(mempool::Request::TransactionIds) + .await + .respond(mempool::Response::TransactionIds( + [tx_id(2)].into_iter().collect(), + )); + }); + + let (rebuild, waited) = wait_once(&cache, &mut change_receiver, &mempool).await; + + responder + .await + .expect("the lagged wait should request the mempool's transaction IDs"); + + assert!(rebuild, "the changed mempool should ask for a rebuild"); + assert_eq!( + waited, MEMPOOL_DEBOUNCE, + "a lagging channel over a mempool that no longer holds what the template was built from \ + should end the wait", + ); +} + /// Checks that a subscription taken before a template is published still reports it. /// /// `getblocktemplate` reads the cache, decides the client already has that template, and only then @@ -81,7 +331,7 @@ async fn a_subscription_reports_a_template_published_before_the_wait() { let mut changes = cache.subscribe(); assert!(cache.is_empty(), "nothing is published yet"); - cache.publish(template()); + cache.publish(template(&HashSet::new()), HashSet::new()); tokio::time::timeout(Duration::from_secs(10), changes.changed()) .await @@ -98,7 +348,7 @@ async fn a_subscription_reports_each_later_publish() { let mut changes = cache.subscribe(); for _ in 0..3 { - cache.publish(template()); + cache.publish(template(&HashSet::new()), HashSet::new()); tokio::time::timeout(Duration::from_secs(10), changes.changed()) .await diff --git a/zebrad/src/commands/start.rs b/zebrad/src/commands/start.rs index 67829b38060..0e7028fb765 100644 --- a/zebrad/src/commands/start.rs +++ b/zebrad/src/commands/start.rs @@ -497,7 +497,7 @@ impl StartCmd { let block_template_task_handle = if config.rpc.listen_addr.is_some() || is_internal_miner_enabled { rpc_impl - .spawn_block_template_updater() + .spawn_block_template_updater(mempool_transaction_subscriber.clone()) .inspect(|_| info!("spawned block template updater task")) } else { None