fixes a clippy lint and a concurrency bug

Author: arya2

zebra-rpc/src/methods/get_block_template_rpcs.rs

@@ -970,7 +970,7 @@ where
 
                 self.mined_block_sender
                     .send((block_hash, block_height))
-                    .map_error_with_prefix(0, "failed to send mined block:")?;
+                    .map_error_with_prefix(0, "failed to send mined block")?;
 
                 return Ok(submit_block::Response::Accepted);
             }

zebra-rpc/src/server/error.rs

@@ -70,6 +70,7 @@ pub(crate) trait MapError<T>: Sized {
     fn map_error(self, code: impl Into<ErrorCode>) -> std::result::Result<T, ErrorObjectOwned>;
 
     /// Maps errors to [`jsonrpsee_types::ErrorObjectOwned`] with a prefixed message and a specific error code.
+    #[cfg(feature = "getblocktemplate-rpcs")]
     fn map_error_with_prefix(
         self,
         code: impl Into<ErrorCode>,
@@ -106,6 +107,7 @@ where
         self.map_err(|error| ErrorObject::owned(code.into().code(), error.to_string(), None::<()>))
     }
 
+    #[cfg(feature = "getblocktemplate-rpcs")]
     fn map_error_with_prefix(
         self,
         code: impl Into<ErrorCode>,

zebrad/src/components/inbound/tests/fake_peer_set.rs

@@ -5,7 +5,7 @@ use std::{collections::HashSet, iter, net::SocketAddr, str::FromStr, sync::Arc,
 use futures::FutureExt;
 use tokio::{sync::oneshot, task::JoinHandle, time::timeout};
 use tower::{buffer::Buffer, builder::ServiceBuilder, util::BoxService, Service, ServiceExt};
-use tracing::Span;
+use tracing::{Instrument, Span};
 
 use zebra_chain::{
     amount::Amount,
@@ -978,15 +978,18 @@ async fn setup(
 
     #[cfg(feature = "getblocktemplate-rpcs")]
     let submitblock_channel = SubmitBlockChannel::new();
-    let sync_gossip_task_handle = tokio::spawn(sync::gossip_best_tip_block_hashes(
-        sync_status.clone(),
-        chain_tip_change.clone(),
-        peer_set.clone(),
-        #[cfg(feature = "getblocktemplate-rpcs")]
-        Some(submitblock_channel.receiver()),
-        #[cfg(not(feature = "getblocktemplate-rpcs"))]
-        None,
-    ));
+    let sync_gossip_task_handle = tokio::spawn(
+        sync::gossip_best_tip_block_hashes(
+            sync_status.clone(),
+            chain_tip_change.clone(),
+            peer_set.clone(),
+            #[cfg(feature = "getblocktemplate-rpcs")]
+            Some(submitblock_channel.receiver()),
+            #[cfg(not(feature = "getblocktemplate-rpcs"))]
+            None,
+        )
+        .in_current_span(),
+    );
 
     let tx_gossip_task_handle = tokio::spawn(gossip_mempool_transaction_id(
         transaction_receiver,

zebrad/src/components/inbound/tests/real_peer_set.rs

@@ -796,7 +796,7 @@ async fn setup(
 mod submitblock_test {
     use std::io;
     use std::sync::{Arc, Mutex};
-    use tracing::Level;
+    use tracing::{Instrument, Level};
     use tracing_subscriber::fmt;
 
     use super::*;
@@ -872,12 +872,15 @@ mod submitblock_test {
 
         // Start the block gossip task with a SubmitBlockChannel
         let submitblock_channel = SubmitBlockChannel::new();
-        let gossip_task_handle = tokio::spawn(sync::gossip_best_tip_block_hashes(
-            sync_status.clone(),
-            chain_tip_change,
-            peer_set.clone(),
-            Some(submitblock_channel.receiver()),
-        ));
+        let gossip_task_handle = tokio::spawn(
+            sync::gossip_best_tip_block_hashes(
+                sync_status.clone(),
+                chain_tip_change,
+                peer_set.clone(),
+                Some(submitblock_channel.receiver()),
+            )
+            .in_current_span(),
+        );
 
         // Send a block top the channel
         submitblock_channel

zebrad/src/components/sync/gossip.rs

@@ -6,6 +6,7 @@ use futures::TryFutureExt;
 use thiserror::Error;
 use tokio::sync::watch;
 use tower::{timeout::Timeout, Service, ServiceExt};
+use tracing::Instrument;
 
 use zebra_chain::block;
 use zebra_network as zn;
@@ -45,7 +46,7 @@ pub enum BlockGossipError {
 ///
 /// [`block::Hash`]: zebra_chain::block::Hash
 pub async fn gossip_best_tip_block_hashes<ZN>(
-    mut sync_status: SyncStatus,
+    sync_status: SyncStatus,
     mut chain_state: ChainTipChange,
     broadcast_network: ZN,
     mut mined_block_receiver: Option<watch::Receiver<(block::Hash, block::Height)>>,
@@ -61,40 +62,56 @@ where
     let mut broadcast_network = Timeout::new(broadcast_network, TIPS_RESPONSE_TIMEOUT);
 
     loop {
-        // wait for at least one tip change, to make sure we have a new block hash to broadcast
-        let tip_action = chain_state.wait_for_tip_change().await.map_err(TipChange)?;
-        // wait until we're close to the tip, because broadcasts are only useful for nodes near the tip
-        // (if they're a long way from the tip, they use the syncer and block locators), unless a mined block
-        // hash is received before `wait_until_close_to_tip()` is ready.
-        let close_to_tip_fut = sync_status.wait_until_close_to_tip().map_err(SyncStatus);
-        let (hash, height) = if let Some(mined_block_receiver) = &mut mined_block_receiver {
+        let mut sync_status = sync_status.clone();
+        let mut chain_tip = chain_state.clone();
+        let tip_change_close_to_network_tip_fut = async move {
+            // wait for at least one tip change, to make sure we have a new block hash to broadcast
+            let tip_action = chain_tip.wait_for_tip_change().await.map_err(TipChange)?;
+
+            // wait until we're close to the tip, because broadcasts are only useful for nodes near the tip
+            // (if they're a long way from the tip, they use the syncer and block locators), unless a mined block
+            // hash is received before `wait_until_close_to_tip()` is ready.
+            sync_status
+                .wait_until_close_to_tip()
+                .map_err(SyncStatus)
+                .await?;
+
+            // get the latest tip change when close to tip - it might be different to the change we awaited,
+            // because the syncer might take a long time to reach the tip
+            let best_tip = chain_tip
+                .last_tip_change()
+                .unwrap_or(tip_action)
+                .best_tip_hash_and_height();
+
+            Ok((best_tip, "sending committed block broadcast", chain_tip))
+        }
+        .in_current_span();
+
+        let ((hash, height), log_msg, updated_chain_state) = if let Some(mined_block_receiver) =
+            mined_block_receiver.as_mut()
+        {
             tokio::select! {
-                close_to_tip = close_to_tip_fut => {
-                    close_to_tip?;
-                    // get the latest tip change - it might be different to the change we awaited,
-                    // because the syncer might take a long time to reach the tip
-                    chain_state.last_tip_change().unwrap_or(tip_action).best_tip_hash_and_height()
+                tip_change_close_to_network_tip = tip_change_close_to_network_tip_fut => {
+                    mined_block_receiver.mark_unchanged();
+                    tip_change_close_to_network_tip?
                 },
 
-                recv_result = mined_block_receiver.changed() => {
-                    recv_result.map_err(SyncStatus)?;
+                Ok(_) = mined_block_receiver.changed() => {
                     // we have a new block to broadcast from the `submitblock `RPC method, get block data and release the channel.
-                    *mined_block_receiver.borrow_and_update()
+                   (*mined_block_receiver.borrow_and_update(), "sending mined block broadcast", chain_state)
                 }
             }
         } else {
-            close_to_tip_fut.await?;
-            chain_state
-                .last_tip_change()
-                .unwrap_or(tip_action)
-                .best_tip_hash_and_height()
+            tip_change_close_to_network_tip_fut.await?
         };
 
+        chain_state = updated_chain_state;
+
         // block broadcasts inform other nodes about new blocks,
         // so our internal Grow or Reset state doesn't matter to them
         let request = zn::Request::AdvertiseBlock(hash);
 
-        debug!(?height, ?request, "sending committed block broadcast");
+        info!(?height, ?request, log_msg);
         // broadcast requests don't return errors, and we'd just want to ignore them anyway
         let _ = broadcast_network
             .ready()