diff --git a/CHANGELOG.md b/CHANGELOG.md index 1db7ffeaac0..b38ae2c84c1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -44,6 +44,15 @@ and this project adheres to [Semantic Versioning](https://semver.org). - Fixed a near-tip sync restart loop when a timed-out `AwaitUtxo` lookup in the transaction verifier was converted to `InternalDowncastError` instead of a missing transparent input. +- Fixed Zakura nodes gossiping and following side-chain blocks. An inbound + `NewBlock` that committed as a side chain (for example a testnet + min-difficulty branch) still advanced the node's Zakura header and verified + frontiers and was forwarded to peers, so a whole fleet could advertise and + propagate a losing branch over Zakura while each node's own chain stayed on + the best one — stranding zakura-only peers that followed the gossip. An + accepted `NewBlock` is now checked against the best chain before it advances + any frontier or is forwarded; side-chain commits are remembered for dedup + only and counted in `sync.header.tip.new_block.side_chain`. - Fixed Zakura header sync permanently stranding a node whose committed chain suffix ends up on a branch the network abandoned. Header sync only requested headers forward from its own frontier hash and treated every non-linking diff --git a/zebra-network/src/zakura/header_sync/events.rs b/zebra-network/src/zakura/header_sync/events.rs index 25b2d2a7b13..78b9b866d84 100644 --- a/zebra-network/src/zakura/header_sync/events.rs +++ b/zebra-network/src/zakura/header_sync/events.rs @@ -185,6 +185,20 @@ pub enum HeaderSyncEvent { /// Duplicate block hash. hash: block::Hash, }, + /// The node's block pipeline accepted an inbound `NewBlock` body, but it + /// committed to a side chain instead of the best chain. The block is + /// remembered for dedup only: a side-chain block must not advance the + /// header or verified frontiers and must not be forwarded to peers, or + /// the whole Zakura layer gossips a losing branch while the node's own + /// chain stays on the best one. + NewBlockAcceptedSideChain { + /// Source peer. + peer: ZakuraPeerId, + /// Accepted side-chain block height. + height: block::Height, + /// Accepted side-chain block hash. + hash: block::Hash, + }, /// The node's block pipeline rejected an inbound `NewBlock` body. NewBlockRejected { /// Source peer. @@ -284,6 +298,7 @@ impl HeaderSyncEvent { Self::FullBlockCommitted { .. } => "full_block_committed", Self::NewBlockAccepted { .. } => "new_block_accepted", Self::NewBlockDuplicate { .. } => "new_block_duplicate", + Self::NewBlockAcceptedSideChain { .. } => "new_block_accepted_side_chain", Self::NewBlockRejected { .. } => "new_block_rejected", Self::WireMessage { .. } => "wire_message", Self::WireDecodeFailed { .. } => "wire_decode_failed", diff --git a/zebra-network/src/zakura/header_sync/reactor.rs b/zebra-network/src/zakura/header_sync/reactor.rs index 85b240908e8..0219ed1f429 100644 --- a/zebra-network/src/zakura/header_sync/reactor.rs +++ b/zebra-network/src/zakura/header_sync/reactor.rs @@ -171,6 +171,9 @@ impl HeaderSyncReactor { HeaderSyncEvent::NewBlockDuplicate { peer, height, hash } => { self.handle_new_block_duplicate(peer, height, hash) } + HeaderSyncEvent::NewBlockAcceptedSideChain { peer, height, hash } => { + self.handle_new_block_accepted_side_chain(peer, height, hash) + } HeaderSyncEvent::NewBlockRejected { peer, hash } => { self.handle_new_block_rejected(peer, hash).await } @@ -556,6 +559,21 @@ impl HeaderSyncReactor { self.trace_new_block_deduped(&peer, height, hash, "already_in_chain"); } + /// Remembers an accepted side-chain `NewBlock` for dedup without advancing + /// any frontier or forwarding it. See + /// [`HeaderSyncEvent::NewBlockAcceptedSideChain`]. + fn handle_new_block_accepted_side_chain( + &mut self, + peer: ZakuraPeerId, + height: block::Height, + hash: block::Hash, + ) { + self.state.pending_new_blocks.remove(&hash); + let _ = self.state.seen.insert(hash); + metrics::counter!("sync.header.tip.new_block.side_chain").increment(1); + self.trace_new_block_deduped(&peer, height, hash, "side_chain"); + } + async fn handle_new_block_rejected(&mut self, peer: ZakuraPeerId, hash: block::Hash) { self.state.pending_new_blocks.remove(&hash); metrics::counter!("sync.header.tip.new_block.rejected").increment(1); @@ -1775,6 +1793,12 @@ impl HeaderSyncReactor { insert_height(row, hs_trace::HEIGHT, *height); insert_hash(row, hs_trace::HASH, *hash); } + HeaderSyncEvent::NewBlockAcceptedSideChain { peer, height, hash } => { + insert_optional_str(row, hs_trace::KIND, Some("new_block_accepted_side_chain")); + insert_peer(row, hs_trace::PEER, peer); + insert_height(row, hs_trace::HEIGHT, *height); + insert_hash(row, hs_trace::HASH, *hash); + } HeaderSyncEvent::NewBlockRejected { peer, hash } => { insert_optional_str(row, hs_trace::KIND, Some("new_block_rejected")); insert_peer(row, hs_trace::PEER, peer); diff --git a/zebra-network/src/zakura/header_sync/tests.rs b/zebra-network/src/zakura/header_sync/tests.rs index ba56cb0c20a..b40079532a2 100644 --- a/zebra-network/src/zakura/header_sync/tests.rs +++ b/zebra-network/src/zakura/header_sync/tests.rs @@ -2903,6 +2903,94 @@ async fn inbound_unseen_valid_new_block_is_seen_and_forwarded_to_eligible_peers( } } +#[tokio::test(flavor = "current_thread")] +async fn accepted_side_chain_new_block_is_deduped_without_advancing_or_forwarding() { + let network = Network::Mainnet; + let block = mainnet_block(&BLOCK_MAINNET_1_BYTES); + let hash = block.hash(); + let height = block.coinbase_height().expect("test block has height"); + let anchor = (block::Height(0), network.genesis_hash()); + let mut fixture = spawn_test_reactor(startup_for(network.clone(), anchor, None)); + let mut tip = fixture.handle.subscribe_tip(); + let source = peer(55); + let would_be_destination = peer(56); + + // The destination's advertised tip is below the block height, so a + // best-chain accept at this height WOULD forward to it. + for peer_id in [source.clone(), would_be_destination.clone()] { + connect_peer(&fixture, peer_id.clone()).await; + advertise_tip( + &fixture, + peer_id, + block::Height(0), + block::Height(0), + DEFAULT_HS_RANGE, + 1, + ) + .await; + } + + fixture + .handle + .send(HeaderSyncEvent::NewBlockAcceptedSideChain { + peer: source.clone(), + height, + hash, + }) + .await + .unwrap(); + + // A side-chain accept advances no frontier and forwards nothing. + while let Ok(Some(action)) = tokio::time::timeout( + std::time::Duration::from_millis(200), + fixture.actions.recv(), + ) + .await + { + if matches!( + action, + HeaderSyncAction::ForwardNewBlock { .. } + | HeaderSyncAction::HeaderAdvanced { .. } + | HeaderSyncAction::HeaderReanchored { .. } + ) { + panic!("side-chain accept must not advance frontiers or forward: {action:?}"); + } + } + assert_eq!(fixture.handle.best_header_tip(), anchor); + assert!( + tokio::time::timeout(std::time::Duration::from_millis(50), tip.changed()) + .await + .is_err(), + "side-chain accept must not publish a new best header tip" + ); + + // The hash is remembered: a later wire NewBlock for it dedups without + // re-entering the block pipeline or scoring the sender. + fixture + .handle + .send(HeaderSyncEvent::WireMessage { + peer: source, + msg: HeaderSyncMessage::NewBlock(block), + }) + .await + .unwrap(); + while let Ok(Some(action)) = tokio::time::timeout( + std::time::Duration::from_millis(200), + fixture.actions.recv(), + ) + .await + { + if matches!( + action, + HeaderSyncAction::NewBlockReceived { .. } + | HeaderSyncAction::ForwardNewBlock { .. } + | HeaderSyncAction::Misbehavior { .. } + ) { + panic!("seen side-chain block must be cheap-deduped without scoring: {action:?}"); + } + } +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn concurrent_duplicate_new_block_dedups_pending_acceptance_without_scoring() { let network = Network::Mainnet; diff --git a/zebrad/src/commands/start.rs b/zebrad/src/commands/start.rs index d3a6799870d..6e20001c166 100644 --- a/zebrad/src/commands/start.rs +++ b/zebrad/src/commands/start.rs @@ -2775,6 +2775,141 @@ mod zakura_header_sync_driver_tests { endpoint.shutdown().await; } + /// End-to-end driver + reactor: an accepted `NewBlock` that committed to a + /// side chain (state `Depth` = `None`) must not advance the header + /// frontier, while a best-chain accept (`Depth` = `Some`) must. + #[tokio::test] + async fn new_block_side_chain_commit_does_not_advance_header_frontier() { + let network = zebra_chain::parameters::Network::Mainnet; + let genesis_hash = network.genesis_hash(); + let mut config = zebra_network::Config { + network: network.clone(), + ..zebra_network::Config::default() + }; + config.zakura.listen_addr = None; + let endpoint = zebra_network::zakura::spawn_zakura_endpoint_with_header_sync_driver( + &config, + |_supervisor, _trace| Arc::new(NoopZakuraService) as Arc, + Some(ZakuraHeaderSyncDriverStartup { + frontiers: HeaderSyncFrontiers { + finalized_height: block::Height(0), + verified_block_tip: block::Height(0), + verified_block_hash: genesis_hash, + }, + best_header_tip: Some((block::Height(0), genesis_hash)), + verified_block_tip_hash: genesis_hash, + }), + ) + .await + .expect("Zakura endpoint starts") + .expect("v2_p2p starts an endpoint"); + let header_sync = endpoint + .header_sync() + .expect("driver startup starts header sync"); + + // Block 2 plays the side-chain commit; block 1 plays the best-chain one. + let side_chain_block = mainnet_block(&BLOCK_MAINNET_2_BYTES); + let side_chain_hash = side_chain_block.hash(); + let best_chain_block = mainnet_block(&BLOCK_MAINNET_1_BYTES); + let best_chain_hash = best_chain_block.hash(); + + let (action_tx, action_rx) = mpsc::channel(4); + let (shutdown_tx, shutdown_rx) = oneshot::channel(); + let handles = ZakuraHeaderSyncDriverHandles { + endpoint: endpoint.clone(), + header_sync: header_sync.clone(), + }; + let state = service_fn(|request: zebra_state::Request| async move { + panic!("unexpected state request from NewBlockReceived: {request:?}"); + #[allow(unreachable_code)] + Ok::<_, zebra_state::BoxError>(zebra_state::Response::Committed(block::Hash([0; 32]))) + }); + let read_state = service_fn(move |request: zebra_state::ReadRequest| async move { + match request { + zebra_state::ReadRequest::Depth(hash) if hash == side_chain_hash => { + Ok::<_, zebra_state::BoxError>(zebra_state::ReadResponse::Depth(None)) + } + zebra_state::ReadRequest::Depth(hash) if hash == best_chain_hash => { + Ok(zebra_state::ReadResponse::Depth(Some(0))) + } + request => panic!("unexpected read request: {request:?}"), + } + }); + // The verifier accepts both blocks; only the state decides which chain + // they landed on. + let verifier = service_fn(|request: zebra_consensus::Request| async move { + match request { + zebra_consensus::Request::Commit(block) => { + Ok::<_, zebra_consensus::BoxError>(block.hash()) + } + request => panic!("unexpected verifier request: {request:?}"), + } + }); + let driver = tokio::spawn(drive_zakura_header_sync_actions( + action_rx, + handles, + state, + read_state, + verifier, + zebra_network::zakura::ZakuraTrace::noop(), + async move { + let _ = shutdown_rx.await; + }, + )); + + let source = + zebra_network::zakura::ZakuraPeerId::new(vec![9; 32]).expect("test peer id is valid"); + action_tx + .send(zebra_network::zakura::HeaderSyncAction::NewBlockReceived { + peer: source.clone(), + height: side_chain_block + .coinbase_height() + .expect("test block has height"), + hash: side_chain_hash, + block: side_chain_block, + }) + .await + .expect("driver action channel stays open"); + + // The side-chain accept must not move the reactor's best header tip. + // Give the driver + reactor time to (incorrectly) advance before + // checking; the follow-up best-chain accept below proves liveness. + tokio::time::sleep(Duration::from_millis(200)).await; + assert_eq!( + header_sync.best_header_tip(), + (block::Height(0), genesis_hash), + "a side-chain NewBlock commit must not advance the header frontier" + ); + + let best_height = best_chain_block + .coinbase_height() + .expect("test block has height"); + action_tx + .send(zebra_network::zakura::HeaderSyncAction::NewBlockReceived { + peer: source, + height: best_height, + hash: best_chain_hash, + block: best_chain_block, + }) + .await + .expect("driver action channel stays open"); + + tokio::time::timeout(Duration::from_secs(2), async { + loop { + if header_sync.best_header_tip() == (best_height, best_chain_hash) { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("a best-chain NewBlock commit advances the header frontier"); + + let _ = shutdown_tx.send(()); + driver.await.expect("driver task exits cleanly"); + endpoint.shutdown().await; + } + #[tokio::test] async fn block_sync_driver_coalesces_stale_needed_queries() { let (action_tx, mut action_rx) = mpsc::channel(8); diff --git a/zebrad/src/commands/start/zakura/header_sync_driver.rs b/zebrad/src/commands/start/zakura/header_sync_driver.rs index d5af4e3c580..3f69c5f641a 100644 --- a/zebrad/src/commands/start/zakura/header_sync_driver.rs +++ b/zebrad/src/commands/start/zakura/header_sync_driver.rs @@ -252,32 +252,58 @@ pub(crate) async fn drive_zakura_header_sync_actions { + // A contextually valid block also commits when it lands + // on a side chain, but only a best-chain block may + // advance the header/verified frontiers or be forwarded + // to peers: gossiping side-chain blocks makes the whole + // Zakura layer follow a losing branch while the node's + // own chain stays honest, stranding zakura-only peers. + let on_best_chain = + new_block_is_on_best_chain(read_state.clone(), hash).await; + let result_label = if on_best_chain { + "accepted" + } else { + "accepted_side_chain" + }; trace_header_commit_finish( &trace, "new_block", &peer, height, hash, - "accepted", + result_label, started, ); trace_header_reactor_event( &trace, - "new_block_accepted", + if on_best_chain { + "new_block_accepted" + } else { + "new_block_accepted_side_chain" + }, Some(&peer), height, hash, 1, ); - let _ = handles - .header_sync - .send(HeaderSyncEvent::NewBlockAccepted { + let event = if on_best_chain { + HeaderSyncEvent::NewBlockAccepted { peer, height, hash, block, - }) - .await; + } + } else { + debug!( + ?peer, + ?height, + ?hash, + "Zakura NewBlock committed to a side chain; \ + not advancing frontiers or forwarding" + ); + HeaderSyncEvent::NewBlockAcceptedSideChain { peer, height, hash } + }; + let _ = handles.header_sync.send(event).await; } Ok(committed_hash) => { trace_header_commit_finish( @@ -1146,6 +1172,44 @@ async fn log_missing_block_bodies( } } +/// Returns whether a just-committed `NewBlock` landed on the best chain. +/// +/// `ReadRequest::Depth` returns `Some` only for best-chain blocks, so it +/// distinguishes a best-chain extension (or a reorg the block just won) from a +/// side-chain commit. Read failures are treated as *not* best-chain: the +/// node's own frontier still advances through the chain-tip mirror, so the +/// only cost of a false negative is skipping one gossip forward, while a +/// false positive would gossip a possibly losing branch. +async fn new_block_is_on_best_chain(read_state: ReadState, hash: block::Hash) -> bool +where + ReadState: Service< + zebra_state::ReadRequest, + Response = zebra_state::ReadResponse, + Error = zebra_state::BoxError, + > + Send + + 'static, + ReadState::Future: Send + 'static, +{ + match read_state + .oneshot(zebra_state::ReadRequest::Depth(hash)) + .await + { + Ok(zebra_state::ReadResponse::Depth(depth)) => depth.is_some(), + Ok(response) => { + warn!(?response, "unexpected Depth response for Zakura NewBlock"); + false + } + Err(error) => { + warn!( + ?hash, + ?error, + "failed to read Zakura NewBlock depth from state" + ); + false + } + } +} + /// Drops the committed body suffix stranded by a header-chain reorg. /// /// `reorged_at` is the first height where a committed header range replaced a