change(rpc): Return from long polling immediately when the chain tip changes

zebra-chain/src/chain_tip.rs

@@ -1,10 +1,11 @@
 //! Zebra interfaces for access to chain tip information.
 
-use std::sync::Arc;
+use std::{future, sync::Arc};
 
 use chrono::{DateTime, Utc};
+use futures::{future::BoxFuture, Future, FutureExt};
 
-use crate::{block, parameters::Network, transaction};
+use crate::{block, parameters::Network, transaction, BoxError};
 
 mod network_chain_tip_height_estimator;
 
@@ -18,32 +19,65 @@ use network_chain_tip_height_estimator::NetworkChainTipHeightEstimator;
 /// An interface for querying the chain tip.
 ///
 /// This trait helps avoid dependencies between:
-/// * zebra-chain and tokio
-/// * zebra-network and zebra-state
+/// * `zebra-chain` and `tokio`
+/// * `zebra-network` and `zebra-state`
 pub trait ChainTip {
-    /// Return the height of the best chain tip.
+    /// Returns the height of the best chain tip.
+    ///
+    /// Does not mark the best tip as seen.
     fn best_tip_height(&self) -> Option<block::Height>;
 
-    /// Return the block hash of the best chain tip.
+    /// Returns the block hash of the best chain tip.
+    ///
+    /// Does not mark the best tip as seen.
     fn best_tip_hash(&self) -> Option<block::Hash>;
 
-    /// Return the height and the hash of the best chain tip.
+    /// Returns the height and the hash of the best chain tip.
+    ///
+    /// Does not mark the best tip as seen.
     fn best_tip_height_and_hash(&self) -> Option<(block::Height, block::Hash)>;
 
-    /// Return the block time of the best chain tip.
+    /// Returns the block time of the best chain tip.
+    ///
+    /// Does not mark the best tip as seen.
     fn best_tip_block_time(&self) -> Option<DateTime<Utc>>;
 
-    /// Return the height and the block time of the best chain tip.
-    ///
+    /// Returns the height and the block time of the best chain tip.
     /// Returning both values at the same time guarantees that they refer to the same chain tip.
+    ///
+    /// Does not mark the best tip as seen.
     fn best_tip_height_and_block_time(&self) -> Option<(block::Height, DateTime<Utc>)>;
 
-    /// Return the mined transaction IDs of the transactions in the best chain tip block.
+    /// Returns the mined transaction IDs of the transactions in the best chain tip block.
     ///
     /// All transactions with these mined IDs should be rejected from the mempool,
     /// even if their authorizing data is different.
+    ///
+    /// Does not mark the best tip as seen.
     fn best_tip_mined_transaction_ids(&self) -> Arc<[transaction::Hash]>;
 
+    /// A future that returns when the best chain tip changes.
+    /// Can return immediately if the latest value in this [`ChainTip`] has not been seen yet.
+    ///
+    /// Marks the best tip as seen.
+    ///
+    /// Returns an error if Zebra is shutting down, or the state has permanently failed.
+    ///
+    /// See [`tokio::watch::Receiver::changed()`](https://docs.rs/tokio/latest/tokio/sync/watch/struct.Receiver.html#method.changed) for details.
+    //
+    // TODO:
+    // Use async_fn_in_trait or return_position_impl_trait_in_trait when one of them stabilises:
+    // https://github.com/rust-lang/rust/issues/91611
+    fn best_tip_changed(&mut self) -> BestTipChanged;
+
+    /// Mark the current best tip as seen.
+    ///
+    /// Later calls to [`ChainTip::best_tip_changed()`] will wait for the next change
+    /// before returning.
+    fn mark_best_tip_seen(&mut self);
+
+    // Provided methods
+    //
     /// Return an estimate of the network chain tip's height.
     ///
     /// The estimate is calculated based on the current local time, the block time of the best tip
@@ -84,7 +118,34 @@ pub trait ChainTip {
     }
 }
 
-/// A chain tip that is always empty.
+/// A future for the [`ChainTip::best_tip_changed()`] method.
+/// See that method for details.
+pub struct BestTipChanged<'f> {
+    fut: BoxFuture<'f, Result<(), BoxError>>,
+}
+
+impl<'f> BestTipChanged<'f> {
+    /// Returns a new [`BestTipChanged`] containing `fut`.
+    pub fn new<Fut>(fut: Fut) -> Self
+    where
+        Fut: Future<Output = Result<(), BoxError>> + Send + 'f,
+    {
+        Self { fut: Box::pin(fut) }
+    }
+}
+
+impl<'f> Future for BestTipChanged<'f> {
+    type Output = Result<(), BoxError>;
+
+    fn poll(
+        mut self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Self::Output> {
+        self.fut.poll_unpin(cx)
+    }
+}
+
+/// A chain tip that is always empty and never changes.
 ///
 /// Used in production for isolated network connections,
 /// and as a mock chain tip in tests.
@@ -115,4 +176,12 @@ impl ChainTip for NoChainTip {
     fn best_tip_mined_transaction_ids(&self) -> Arc<[transaction::Hash]> {
         Arc::new([])
     }
+
+    /// The [`NoChainTip`] best tip never changes, so this never returns.
+    fn best_tip_changed(&mut self) -> BestTipChanged {
+        BestTipChanged::new(future::pending())
+    }
+
+    /// The [`NoChainTip`] best tip never changes, so this does nothing.
+    fn mark_best_tip_seen(&mut self) {}
 }

zebra-chain/src/chain_tip/mock.rs

@@ -3,11 +3,19 @@
 use std::sync::Arc;
 
 use chrono::{DateTime, Utc};
+use futures::{future, FutureExt, TryFutureExt};
 use tokio::sync::watch;
 
-use crate::{block, chain_tip::ChainTip, parameters::Network, transaction};
+use crate::{
+    block,
+    chain_tip::{BestTipChanged, ChainTip},
+    parameters::Network,
+    transaction,
+};
 
 /// A sender to sets the values read by a [`MockChainTip`].
+//
+// Update `best_tip_changed()` for each new field that is added to MockChainTipSender.
 pub struct MockChainTipSender {
     /// A sender that sets the `best_tip_height` of a [`MockChainTip`].
     best_tip_height: watch::Sender<Option<block::Height>>,
@@ -112,6 +120,40 @@ impl ChainTip for MockChainTip {
                     .map(|tip_height| (estimated_distance, tip_height))
             })
     }
+
+    /// Returns when any sender channel changes.
+    /// Returns an error if any sender was dropped.
+    ///
+    /// Marks the changed channel as seen when the returned future completes.
+    //
+    // Update this method when each new mock field is added.
+    fn best_tip_changed(&mut self) -> BestTipChanged {
+        // A future that returns when the first watch channel has changed
+        let select_changed = future::select_all([
+            // Erase the differing future types for each channel, and map their error types
+            BestTipChanged::new(self.best_tip_height.changed().err_into()),
+            BestTipChanged::new(self.best_tip_hash.changed().err_into()),
+            BestTipChanged::new(self.best_tip_block_time.changed().err_into()),
+            BestTipChanged::new(
+                self.estimated_distance_to_network_chain_tip
+                    .changed()
+                    .err_into(),
+            ),
+        ])
+        // Map the select result to the expected type, dropping the unused channels
+        .map(|(changed_result, _changed_index, _remaining_futures)| changed_result);
+
+        BestTipChanged::new(select_changed)
+    }
+
+    /// Marks all sender channels as seen.
+    fn mark_best_tip_seen(&mut self) {
+        self.best_tip_height.borrow_and_update();
+        self.best_tip_hash.borrow_and_update();
+        self.best_tip_block_time.borrow_and_update();
+        self.estimated_distance_to_network_chain_tip
+            .borrow_and_update();
+    }
 }
 
 impl MockChainTipSender {

zebra-rpc/Cargo.toml

@@ -70,11 +70,12 @@ zebra-script = { path = "../zebra-script" }
 zebra-state = { path = "../zebra-state" }
 
 [dev-dependencies]
-insta = { version = "1.23.0", features = ["redactions", "json"] }
+insta = { version = "1.23.0", features = ["redactions", "json", "ron"] }
+
 proptest = "0.10.1"
 proptest-derive = "0.3.0"
-thiserror = "1.0.37"
 
+thiserror = "1.0.37"
 tokio = { version = "1.23.0", features = ["full", "tracing", "test-util"] }
 
 zebra-chain = { path = "../zebra-chain", features = ["proptest-impl"] }