Add a ChainTipChange type to await chain tip changes

zebra-chain/src/chain_tip.rs

@@ -1,4 +1,4 @@
-//! Chain tip interfaces.
+//! Zebra interfaces for access to chain tip information.
 
 use std::sync::Arc;
 

zebra-network/src/isolated.rs

@@ -58,7 +58,7 @@ pub fn connect_isolated(
             Ok::<Response, Box<dyn std::error::Error + Send + Sync + 'static>>(Response::Nil)
         }))
         .with_user_agent(user_agent)
-        .with_chain_tip_receiver(NoChainTip)
+        .with_latest_chain_tip(NoChainTip)
         .finish()
         .expect("provided mandatory builder parameters");
 

zebra-network/src/peer/handshake.rs

@@ -57,7 +57,7 @@ pub struct Handshake<S, C = NoChainTip> {
     our_services: PeerServices,
     relay: bool,
     parent_span: Span,
-    chain_tip_receiver: C,
+    latest_chain_tip: C,
 }
 
 /// The peer address that we are handshaking with.
@@ -307,7 +307,7 @@ pub struct Builder<S, C = NoChainTip> {
     user_agent: Option<String>,
     relay: Option<bool>,
     inv_collector: Option<broadcast::Sender<(InventoryHash, SocketAddr)>>,
-    chain_tip_receiver: C,
+    latest_chain_tip: C,
 }
 
 impl<S, C> Builder<S, C>
@@ -374,9 +374,9 @@ where
     /// constant over network upgrade activations.
     ///
     /// Use [`NoChainTip`] to explicitly provide no chain tip.
-    pub fn with_chain_tip_receiver<NewC>(self, chain_tip_receiver: NewC) -> Builder<S, NewC> {
+    pub fn with_latest_chain_tip<NewC>(self, latest_chain_tip: NewC) -> Builder<S, NewC> {
         Builder {
-            chain_tip_receiver,
+            latest_chain_tip,
             // TODO: Until Rust RFC 2528 reaches stable, we can't do `..self`
             config: self.config,
             inbound_service: self.inbound_service,
@@ -429,7 +429,7 @@ where
             our_services,
             relay,
             parent_span: Span::current(),
-            chain_tip_receiver: self.chain_tip_receiver,
+            latest_chain_tip: self.latest_chain_tip,
         })
     }
 }
@@ -452,7 +452,7 @@ where
             our_services: None,
             relay: None,
             inv_collector: None,
-            chain_tip_receiver: NoChainTip,
+            latest_chain_tip: NoChainTip,
         }
     }
 }
@@ -471,7 +471,7 @@ pub async fn negotiate_version(
     user_agent: String,
     our_services: PeerServices,
     relay: bool,
-    chain_tip_receiver: impl ChainTip,
+    latest_chain_tip: impl ChainTip,
 ) -> Result<(Version, PeerServices, SocketAddr), HandshakeError> {
     // Create a random nonce for this connection
     let local_nonce = Nonce::default();
@@ -585,7 +585,7 @@ pub async fn negotiate_version(
 
     // SECURITY: Reject connections to peers on old versions, because they might not know about all
     // network upgrades and could lead to chain forks or slower block propagation.
-    let height = chain_tip_receiver.best_tip_height();
+    let height = latest_chain_tip.best_tip_height();
     let min_version = Version::min_remote_for_height(config.network, height);
     if remote_version < min_version {
         // Disconnect if peer is using an obsolete version.
@@ -643,7 +643,7 @@ where
         let user_agent = self.user_agent.clone();
         let our_services = self.our_services;
         let relay = self.relay;
-        let chain_tip_receiver = self.chain_tip_receiver.clone();
+        let latest_chain_tip = self.latest_chain_tip.clone();
 
         let fut = async move {
             debug!(
@@ -674,7 +674,7 @@ where
                     user_agent,
                     our_services,
                     relay,
-                    chain_tip_receiver,
+                    latest_chain_tip,
                 ),
             )
             .await??;

zebra-network/src/peer_set/initialize.rs

@@ -37,7 +37,7 @@ mod tests;
 type PeerChange = Result<Change<SocketAddr, peer::Client>, BoxError>;
 
 /// Initialize a peer set, using a network `config`, `inbound_service`,
-/// and `chain_tip_receiver`.
+/// and `latest_chain_tip`.
 ///
 /// The peer set abstracts away peer management to provide a
 /// [`tower::Service`] representing "the network" that load-balances requests
@@ -62,7 +62,7 @@ type PeerChange = Result<Change<SocketAddr, peer::Client>, BoxError>;
 pub async fn init<S, C>(
     config: Config,
     inbound_service: S,
-    chain_tip_receiver: C,
+    latest_chain_tip: C,
 ) -> (
     Buffer<BoxService<Request, Response, BoxError>, Request>,
     Arc<std::sync::Mutex<AddressBook>>,
@@ -92,7 +92,7 @@ where
             .with_timestamp_collector(timestamp_collector)
             .with_advertised_services(PeerServices::NODE_NETWORK)
             .with_user_agent(crate::constants::USER_AGENT.to_string())
-            .with_chain_tip_receiver(chain_tip_receiver)
+            .with_latest_chain_tip(latest_chain_tip)
             .want_transactions(true)
             .finish()
             .expect("configured all required parameters");

zebra-state/src/arbitrary.rs

@@ -8,7 +8,7 @@ use zebra_chain::{
     value_balance::ValueBalance,
 };
 
-use crate::{request::ContextuallyValidBlock, PreparedBlock};
+use crate::{request::ContextuallyValidBlock, service::chain_tip::ChainTipBlock, PreparedBlock};
 
 /// Mocks computation done during semantic validation
 pub trait Prepare {
@@ -33,6 +33,32 @@ impl Prepare for Arc<Block> {
     }
 }
 
+impl<T> From<T> for ChainTipBlock
+where
+    T: Prepare,
+{
+    fn from(block: T) -> Self {
+        block.prepare().into()
+    }
+}
+
+impl From<PreparedBlock> for ChainTipBlock {
+    fn from(prepared: PreparedBlock) -> Self {
+        let PreparedBlock {
+            block: _,
+            hash,
+            height,
+            new_outputs: _,
+            transaction_hashes,
+        } = prepared;
+        Self {
+            hash,
+            height,
+            transaction_hashes,
+        }
+    }
+}
+
 impl PreparedBlock {
     /// Returns a [`ContextuallyValidBlock`] created from this block,
     /// with fake zero-valued spent UTXOs.

zebra-state/src/lib.rs

@@ -35,7 +35,7 @@ pub use constants::MAX_BLOCK_REORG_HEIGHT;
 pub use error::{BoxError, CloneError, CommitBlockError, ValidateContextError};
 pub use request::{FinalizedBlock, HashOrHeight, PreparedBlock, Request};
 pub use response::Response;
-pub use service::{chain_tip::ChainTipReceiver, init};
+pub use service::{chain_tip::LatestChainTip, init};
 
 #[cfg(any(test, feature = "proptest-impl"))]
 pub use service::init_test;

zebra-state/src/service.rs

@@ -29,7 +29,7 @@ use crate::{
 };
 
 use self::{
-    chain_tip::{ChainTipReceiver, ChainTipSender},
+    chain_tip::{ChainTipChange, ChainTipSender, LatestChainTip},
     non_finalized_state::{NonFinalizedState, QueuedBlocks},
 };
 
@@ -76,13 +76,14 @@ pub(crate) struct StateService {
 impl StateService {
     const PRUNE_INTERVAL: Duration = Duration::from_secs(30);
 
-    pub fn new(config: Config, network: Network) -> (Self, ChainTipReceiver) {
+    pub fn new(config: Config, network: Network) -> (Self, LatestChainTip, ChainTipChange) {
         let disk = FinalizedState::new(&config, network);
         let initial_tip = disk
             .tip_block()
             .map(FinalizedBlock::from)
             .map(ChainTipBlock::from);
-        let (chain_tip_sender, chain_tip_receiver) = ChainTipSender::new(initial_tip);
+        let (chain_tip_sender, latest_chain_tip, chain_tip_change) =
+            ChainTipSender::new(initial_tip);
 
         let mem = NonFinalizedState::new(network);
         let queued_blocks = QueuedBlocks::default();
@@ -122,7 +123,7 @@ impl StateService {
         }
         tracing::info!("no legacy chain found");
 
-        (state, chain_tip_receiver)
+        (state, latest_chain_tip, chain_tip_change)
     }
 
     /// Queue a finalized block for verification and storage in the finalized state.
@@ -769,7 +770,7 @@ impl Service<Request> for StateService {
 }
 
 /// Initialize a state service from the provided [`Config`].
-/// Returns a boxed state service, and a receiver for state chain tip updates.
+/// Returns a boxed state service, and receivers for state chain tip updates.
 ///
 /// Each `network` has its own separate on-disk database.
 ///
@@ -780,18 +781,26 @@ impl Service<Request> for StateService {
 pub fn init(
     config: Config,
     network: Network,
-) -> (BoxService<Request, Response, BoxError>, ChainTipReceiver) {
-    let (state_service, chain_tip_receiver) = StateService::new(config, network);
-
-    (BoxService::new(state_service), chain_tip_receiver)
+) -> (
+    BoxService<Request, Response, BoxError>,
+    LatestChainTip,
+    ChainTipChange,
+) {
+    let (state_service, latest_chain_tip, chain_tip_change) = StateService::new(config, network);
+
+    (
+        BoxService::new(state_service),
+        latest_chain_tip,
+        chain_tip_change,
+    )
 }
 
 /// Initialize a state service with an ephemeral [`Config`] and a buffer with a single slot.
 ///
 /// This can be used to create a state service for testing. See also [`init`].
 #[cfg(any(test, feature = "proptest-impl"))]
 pub fn init_test(network: Network) -> Buffer<BoxService<Request, Response, BoxError>, Request> {
-    let (state_service, _) = StateService::new(Config::ephemeral(), network);
+    let (state_service, _, _) = StateService::new(Config::ephemeral(), network);
 
     Buffer::new(BoxService::new(state_service), 1)
 }