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