feat: Chain follower implemetation

Author: Felipe Rosa

.config/dictionaries/project.dic

@@ -3,10 +3,13 @@ addrr
 adminer
 asyncio
 auditability
+backpressure
+blockfetch
 BROTLI
 cardano
 cbor
 CEST
+chainsync
 chrono
 ciphertext
 codepoints

hermes/crates/cardano-chain-follower/examples/follow_chain_updates.rs

@@ -4,9 +4,19 @@
 use std::error::Error;
 
 use cardano_chain_follower::{ChainUpdate, Follower, FollowerConfigBuilder, Network};
+use tracing::level_filters::LevelFilter;
+use tracing_subscriber::EnvFilter;
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn Error>> {
+    tracing_subscriber::fmt()
+        .with_env_filter(
+            EnvFilter::builder()
+                .with_default_directive(LevelFilter::INFO.into())
+                .from_env_lossy(),
+        )
+        .init();
+
     // Defaults to start following from the tip.
     let config = FollowerConfigBuilder::default().build();
 
@@ -17,7 +27,8 @@ async fn main() -> Result<(), Box<dyn Error>> {
     )
     .await?;
 
-    loop {
+    // Wait for 3 chain updates and shutdown.
+    for _ in 0..3 {
         let chain_update = follower.next().await?;
 
         match chain_update {
@@ -43,4 +54,9 @@ async fn main() -> Result<(), Box<dyn Error>> {
             },
         }
     }
+
+    // Waits for the follower background task to exit.
+    follower.close().await?;
+
+    Ok(())
 }

hermes/crates/cardano-chain-follower/src/lib.rs

@@ -7,15 +7,23 @@
 //         if it's compiled with this flag.
 #![deny(missing_docs)]
 
+use std::sync::Arc;
+
 pub use pallas::network::miniprotocols::Point;
 use pallas::{
     ledger::traverse::MultiEraBlock,
     network::{
         facades::PeerClient,
-        miniprotocols::{MAINNET_MAGIC, PREVIEW_MAGIC, PRE_PRODUCTION_MAGIC, TESTNET_MAGIC},
+        miniprotocols::{
+            chainsync, MAINNET_MAGIC, PREVIEW_MAGIC, PRE_PRODUCTION_MAGIC, TESTNET_MAGIC,
+        },
     },
 };
 use thiserror::Error;
+use tokio::{
+    sync::{mpsc, Mutex},
+    task::JoinHandle,
+};
 
 /// Default [`Follower`] block buffer size.
 const DEFAULT_CHAIN_UPDATE_BUFFER_SIZE: usize = 32;
@@ -36,7 +44,10 @@ pub enum Error {
     Blockfetch(pallas::network::miniprotocols::blockfetch::ClientError),
     /// Chainsync protocol error.
     #[error("Chainsync error: {0:?}")]
-    Chainsync(pallas::network::miniprotocols::chainsync::ClientError),
+    Chainsync(chainsync::ClientError),
+    /// Follower start point was not found.
+    #[error("Follower start point was not found")]
+    FollowerStartPointNotFound,
 }
 
 /// Crate result type.
@@ -131,9 +142,7 @@ impl Reader {
     ///
     /// Returns Err if the block was not found or if some communication error ocurred.
     pub async fn read_block<P>(&mut self, at: P) -> Result<MultiEraBlockData>
-    where
-        P: Into<PointOrTip>,
-    {
+    where P: Into<PointOrTip> {
         let point = self.resolve_point_or_tip(at.into()).await?;
 
         let block_data = self
@@ -160,9 +169,7 @@ impl Reader {
     pub async fn read_block_range<P>(
         &mut self, from: Point, to: P,
     ) -> Result<Vec<MultiEraBlockData>>
-    where
-        P: Into<PointOrTip>,
-    {
+    where P: Into<PointOrTip> {
         let to_point = self.resolve_point_or_tip(to.into()).await?;
 
         let data_vec = self
@@ -178,6 +185,7 @@ impl Reader {
         Ok(data_vec)
     }
 
+    /// Resolves [`PointOrTip`] to a point finding the tip point if needed.
     #[inline]
     async fn resolve_point_or_tip(&mut self, point_or_tip: PointOrTip) -> Result<Point> {
         match point_or_tip {
@@ -254,9 +262,7 @@ impl FollowerConfigBuilder {
     /// * `from`: Sync starting point.
     #[must_use]
     pub fn follow_from<P>(mut self, from: P) -> Self
-    where
-        P: Into<PointOrTip>,
-    {
+    where P: Into<PointOrTip> {
         self.follow_from = from.into();
         self
     }
@@ -283,7 +289,14 @@ pub struct FollowerConfig {
 }
 
 /// Cardano chain follower.
-pub struct Follower {}
+pub struct Follower {
+    /// Client shared by the follower and its task.
+    client: Arc<Mutex<PeerClient>>,
+    /// Chain update receiver.
+    chain_update_rx: mpsc::Receiver<Result<ChainUpdate>>,
+    /// Task thread join handle.
+    task_join_handle: Option<JoinHandle<()>>,
+}
 
 impl Follower {
     /// Connects the follower to a producer using the node-to-node protocol.
@@ -297,14 +310,43 @@ impl Follower {
     /// # Errors
     ///
     /// Returns Err if the connection could not be established.
-    pub async fn connect(
-        _address: &str, _network: Network, _config: FollowerConfig,
-    ) -> Result<Self> {
-        todo!()
+    pub async fn connect(address: &str, network: Network, config: FollowerConfig) -> Result<Self> {
+        let client = Arc::new(Mutex::new(
+            PeerClient::connect(address, network.into())
+                .await
+                .map_err(Error::Client)?,
+        ));
+
+        let (chain_update_tx, chain_update_rx) = mpsc::channel(config.chain_update_buffer_size);
+
+        let mut this = Self {
+            client: client.clone(),
+            chain_update_rx,
+            task_join_handle: None,
+        };
+
+        let start_point = this
+            .set_read_pointer(config.follow_from)
+            .await?
+            .ok_or(Error::FollowerStartPointNotFound)?;
+        tracing::debug!(
+            slot = start_point.slot_or_default(),
+            "Follower read pointer set to starting point"
+        );
+
+        let task_join_handle = tokio::spawn(follow_task::run(
+            client,
+            chain_update_tx,
+            config.max_await_retries,
+        ));
+        this.task_join_handle = Some(task_join_handle);
+
+        Ok(this)
     }
 
-    /// Set the follower's chain read-pointer. Returns None if the point was
-    /// not found on the chain.
+    /// Set the follower's chain read-pointer.
+    ///
+    /// Returns None if the point was not found on the chain.
     ///
     /// # Arguments
     ///
@@ -313,20 +355,194 @@ impl Follower {
     /// # Errors
     ///
     /// Returns Err if something went wrong while communicating with the producer.
-    pub async fn set_read_pointer<P>(&mut self, _at: P) -> Result<Option<Point>>
-    where
-        P: Into<PointOrTip>,
-    {
-        todo!()
+    pub async fn set_read_pointer<P>(&mut self, at: P) -> Result<Option<Point>>
+    where P: Into<PointOrTip> {
+        let mut client = self.client.lock().await;
+
+        match Into::<PointOrTip>::into(at) {
+            PointOrTip::Point(Point::Origin) => {
+                let point = client
+                    .chainsync()
+                    .intersect_origin()
+                    .await
+                    .map_err(Error::Chainsync)?;
+
+                Ok(Some(point))
+            },
+            PointOrTip::Point(p @ Point::Specific(..)) => {
+                client
+                    .chainsync()
+                    .find_intersect(vec![p])
+                    .await
+                    .map(|(point, _)| point)
+                    .map_err(Error::Chainsync)
+            },
+            PointOrTip::Tip => {
+                let point = client
+                    .chainsync()
+                    .intersect_tip()
+                    .await
+                    .map_err(Error::Chainsync)?;
+
+                Ok(Some(point))
+            },
+        }
     }
 
     /// Receive the next chain update from the producer.
     ///
     /// # Errors
     ///
     /// Returns Err if any producer communication errors occurred.
+    #[allow(clippy::missing_panics_doc)]
     pub async fn next(&mut self) -> Result<ChainUpdate> {
-        todo!()
+        // This will not panic
+        #[allow(clippy::expect_used)]
+        self.chain_update_rx
+            .recv()
+            .await
+            .expect("Follow task should be running")
+    }
+
+    /// Closes the follower connection and stops its background task.
+    ///
+    /// # Errors
+    ///
+    /// Returns Err if some error occurred in the background task.
+    pub async fn close(mut self) -> std::result::Result<(), tokio::task::JoinError> {
+        self.chain_update_rx.close();
+
+        if let Some(join_handle) = self.task_join_handle {
+            join_handle.await
+        } else {
+            Ok(())
+        }
+    }
+}
+
+/// Contains functions related to the Follower's background task.
+mod follow_task {
+    use std::sync::Arc;
+
+    use pallas::{
+        ledger::traverse::MultiEraHeader,
+        network::{
+            facades::PeerClient,
+            miniprotocols::{chainsync, Point},
+        },
+    };
+    use tokio::sync::{mpsc, oneshot, Mutex};
+
+    use crate::{ChainUpdate, Error, MultiEraBlockData};
+
+    /// Runs a [`Follower`](super::Follower) background task.
+    ///
+    /// The task runs until the chain update channel is closed (e.g. when the follower is
+    /// dropped or the close fn is called).
+    ///
+    /// It keeps asking the connected node new chain updates. Every update and
+    /// communication errors are sent through the channel to the follower.
+    ///
+    /// Backpressure is achieved with the channel's limited size.
+    pub async fn run(
+        client: Arc<Mutex<PeerClient>>, chain_update_tx: mpsc::Sender<crate::Result<ChainUpdate>>,
+        max_retries_count: u32,
+    ) {
+        'main: loop {
+            let try_count = 0;
+
+            'tries: loop {
+                assert!(try_count <= max_retries_count, "Node misbehavior");
+
+                let (cancel_tx, _cancel_rx) = oneshot::channel::<()>();
+
+                tokio::select! {
+                    () = chain_update_tx.closed() => {
+                        break 'main;
+                    }
+
+                    res = get_next_response(client.clone(), cancel_tx) => match res {
+                        Err(err) => {
+                            if chain_update_tx.send(Err(err)).await.is_err() {
+                                break 'main;
+                            }
+                        },
+                        Ok(next_response) => {
+                            if let Some(chain_update) = next_response {
+                                if chain_update_tx.send(Ok(chain_update)).await.is_err() {
+                                    break 'tries;
+                                }
+                            }
+                        }
+                    }
+                };
+            }
+        }
+
+        tracing::debug!("Follower background task shutdown");
+    }
+
+    /// Waits for the next update from the node the client is connected to.
+    ///
+    /// Can be cancelled by closing the `cancel_tx` receiver end (explicitly or by
+    /// dropping it).
+    async fn get_next_response(
+        client: Arc<Mutex<PeerClient>>, mut cancel_tx: oneshot::Sender<()>,
+    ) -> crate::Result<Option<ChainUpdate>> {
+        let res = {
+            let mut client_lock = client.lock().await;
+
+            if client_lock.chainsync().has_agency() {
+                tokio::select! {
+                    () = cancel_tx.closed() => { return Ok(None); }
+                    res = client_lock.chainsync().request_next() => { res }
+                }
+            } else {
+                tokio::select! {
+                    () = cancel_tx.closed() => { return Ok(None); }
+                    res = client_lock.chainsync().recv_while_must_reply() => { res }
+                }
+            }
+            .map_err(Error::Chainsync)?
+        };
+
+        match res {
+            chainsync::NextResponse::RollForward(header, _tip) => {
+                let decoded_header = MultiEraHeader::decode(
+                    header.variant,
+                    header.byron_prefix.map(|p| p.0),
+                    &header.cbor,
+                )
+                .map_err(Error::Codec)?;
+
+                let mut client_lock = client.lock().await;
+
+                let req_fut = client_lock.blockfetch().fetch_single(Point::Specific(
+                    decoded_header.slot(),
+                    decoded_header.hash().to_vec(),
+                ));
+
+                let block_data = tokio::select! {
+                    () = cancel_tx.closed() => { return Ok(None); }
+                    res = req_fut => { res.map_err(Error::Blockfetch)? }
+                };
+
+                Ok(Some(ChainUpdate::Block(MultiEraBlockData(block_data))))
+            },
+            chainsync::NextResponse::RollBackward(point, _tip) => {
+                let mut client_lock = client.lock().await;
+
+                let req_fut = client_lock.blockfetch().fetch_single(point);
+
+                let block_data = tokio::select! {
+                    () = cancel_tx.closed() => { return Ok(None); }
+                    res = req_fut => { res.map_err(Error::Blockfetch)? }
+                };
+
+                Ok(Some(ChainUpdate::Rollback(MultiEraBlockData(block_data))))
+            },
+            chainsync::NextResponse::Await => Ok(None),
+        }
     }
 }