fix(zakura): only advance and gossip NewBlocks that land on the best chain

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

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",

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);

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;

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<dyn ZakuraService>,
+            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);