Merge branch 'main' into refactor/split-bearer-sync

examples/n2n-miniprotocols/Cargo.toml

@@ -11,6 +11,8 @@ pallas = { path = "../../pallas" }
 net2 = "0.2.37"
 hex = "0.4.3"
 log = "0.4.16"
+thiserror = "1.0.31"
+futures = "0.3.29"
 tracing = "0.1.37"
 tracing-subscriber = "0.3.16"
 tokio = { version = "1.27.0", features = ["rt-multi-thread"] }

examples/n2n-miniprotocols/src/main.rs

@@ -1,77 +1,186 @@
-use pallas::network::{
+use pallas::{network::{
     facades::PeerClient,
-    miniprotocols::{chainsync, Point, MAINNET_MAGIC},
-};
-use tokio::time::Instant;
-use tracing::info;
-
-async fn do_blockfetch(peer: &mut PeerClient) {
-    let range = (
-        Point::Specific(
-            43847831,
-            hex::decode("15b9eeee849dd6386d3770b0745e0450190f7560e5159b1b3ab13b14b2684a45")
-                .unwrap(),
-        ),
-        Point::Specific(
-            43847844,
-            hex::decode("ff8d558a3d5a0e058beb3d94d26a567f75cd7d09ff5485aa0d0ebc38b61378d4")
-                .unwrap(),
-        ),
-    );
-
-    let blocks = peer.blockfetch().fetch_range(range).await.unwrap();
-
-    for block in blocks {
-        info!("received block of size: {}", block.len());
+    miniprotocols::{chainsync, Point, MAINNET_MAGIC, blockfetch, keepalive},
+}, ledger::traverse::MultiEraHeader};
+use tokio::{time::Instant, select};
+use thiserror::Error;
+use futures::{future::FutureExt, pin_mut};
+
+#[derive(Error, Debug)]
+pub enum Error {
+    #[error("hex conversion error")]
+    FromHexError(#[from] hex::FromHexError),
+
+    #[error("blockfetch error")]
+    BlockFetchError(#[from] blockfetch::ClientError),
+
+    #[error("chainsync error")]
+    ChainSyncError(#[from] chainsync::ClientError),
+
+    #[error("keepalive error")]
+    KeepAliveError(#[from] keepalive::Error),
+
+    #[error("pallas_traverse error")]
+    PallasTraverseError(#[from] pallas::ledger::traverse::Error),
+}
+
+async fn do_blockfetch(blockfetch_client: &mut blockfetch::Client, range: (Point, Point)) -> Result<(), Error> {
+    let blocks = blockfetch_client.fetch_range(range.clone()).await?;
+
+    for block in &blocks {
+        tracing::trace!("received block of size: {}", block.len());
     }
+    tracing::info!("received {} blocks. last slot: {}", blocks.len(), range.1.slot_or_default());
+    Ok(())
 }
 
-async fn do_chainsync(peer: &mut PeerClient) {
+async fn do_chainsync(chainsync_client: &mut chainsync::N2NClient, blockfetch_client: &mut blockfetch::Client) -> Result<(), Error> {
     let known_points = vec![Point::Specific(
         43847831u64,
-        hex::decode("15b9eeee849dd6386d3770b0745e0450190f7560e5159b1b3ab13b14b2684a45").unwrap(),
+        hex::decode("15b9eeee849dd6386d3770b0745e0450190f7560e5159b1b3ab13b14b2684a45")?,
     )];
 
-    let (point, _) = peer.chainsync().find_intersect(known_points).await.unwrap();
+    let (point, _) = chainsync_client.find_intersect(known_points).await?;
 
-    info!("intersected point is {:?}", point);
+    tracing::info!("intersected point is {:?}", point);
 
-    let mut keepalive_timer = Instant::now();
-    for _ in 0..10 {
-        if keepalive_timer.elapsed().as_secs() > 20 {
-            peer.keepalive().send_keepalive().await.unwrap();
-            keepalive_timer = Instant::now();
-        }
-        let next = peer.chainsync().request_next().await.unwrap();
+    let mut block_count = 0u16;
+    let mut start_point = Point::Specific(0, vec![]);
+    let mut end_point: Point;
+    let mut next_log = Instant::now();
+    loop {
+        let next = chainsync_client.request_next().await?;
 
         match next {
             chainsync::NextResponse::RollForward(h, _) => {
-                log::info!("rolling forward, header size: {}", h.cbor.len())
+                tracing::trace!("rolling forward, header size: {}", h.cbor.len());
+                let point = match h.byron_prefix {
+                    None => {
+                        let multi_era_header = MultiEraHeader::decode(h.variant, None, &h.cbor)?;
+                        let slot = multi_era_header.slot();
+                        let hash = multi_era_header.hash().to_vec();
+                        let number = multi_era_header.number();
+                        match &multi_era_header {
+                            MultiEraHeader::EpochBoundary(_) => {
+                                tracing::info!("epoch boundary");
+                                None
+                            },
+                            MultiEraHeader::AlonzoCompatible(_) | MultiEraHeader::Babbage(_) => {
+                                if next_log.elapsed().as_secs() > 1 {
+                                    tracing::info!("chainsync block header: {}", number);
+                                    next_log = Instant::now();
+                                }
+                                Some(Point::Specific(slot, hash))
+                            },
+                            MultiEraHeader::Byron(_) => {
+                                tracing::info!("ignoring byron header");
+                                None
+                            },
+                        }
+                    }
+                    Some(_) => {
+                        tracing::info!("skipping byron block");
+                        None
+                    }
+                };
+                match point {
+                    Some(p) => {
+                        block_count += 1;
+                        if block_count == 1 {
+                            start_point = p;
+                        } 
+                        else if block_count == 10 {
+                            end_point = p;
+                            do_blockfetch(blockfetch_client, (start_point.clone(), end_point.clone())).await?;
+                            block_count = 0;
+                        }
+                    },
+                    None => {},
+                };
             }
             chainsync::NextResponse::RollBackward(x, _) => log::info!("rollback to {:?}", x),
-            chainsync::NextResponse::Await => log::info!("tip of chaing reached"),
+            chainsync::NextResponse::Await => tracing::info!("tip of chaing reached"),
         };
     }
 }
 
+async fn do_keepalive(keepalive_client: &mut keepalive::Client) -> Result<(), Error> {
+    let mut keepalive_timer = Instant::now();
+    loop {
+        if keepalive_timer.elapsed().as_secs() > 20 {
+            tracing::info!("sending keepalive...");
+            keepalive_client.send_keepalive().await?;
+            tracing::info!("keepalive sent");
+            keepalive_timer = Instant::now();
+        }
+    }
+}
+
 #[tokio::main]
 async fn main() {
     tracing::subscriber::set_global_default(
         tracing_subscriber::FmtSubscriber::builder()
-            .with_max_level(tracing::Level::TRACE)
+            .with_max_level(tracing::Level::INFO)
             .finish(),
     )
     .unwrap();
 
-    // setup a TCP socket to act as data bearer between our agents and the remote
-    // relay.
-    let mut peer = PeerClient::connect("relays-new.cardano-mainnet.iohk.io:3001", MAINNET_MAGIC)
-        .await
-        .unwrap();
+    loop {
+        // setup a TCP socket to act as data bearer between our agents and the remote
+        // relay.
+        let server = "backbone.cardano-mainnet.iohk.io:3001";
+        // let server = "localhost:6000";
+        let mut peer = PeerClient::connect(server, MAINNET_MAGIC)
+            .await
+            .unwrap();
 
-    // fetch an arbitrary batch of block
-    do_blockfetch(&mut peer).await;
+            let chainsync_handle = tokio::spawn(async move {
+            do_chainsync(&mut peer.chainsync, &mut peer.blockfetch).await?;
+            Ok::<(), Error>(())
+        }).fuse();
+        let keepalive_handle = tokio::spawn(async move {
+            do_keepalive(&mut peer.keepalive).await?;
+            Ok::<(), Error>(())
+        }).fuse();
 
-    // execute the chainsync flow from an arbitrary point in the chain
-    do_chainsync(&mut peer).await;
+        pin_mut!(chainsync_handle, keepalive_handle);
+
+        // If any of these concurrent tasks exit or fail, the others are canceled.
+        select! {
+            chainsync_result = chainsync_handle => {
+                match chainsync_result {
+                    Ok(result) => {
+                        match result {
+                            Ok(_) => {}
+                            Err(error) => {
+                                tracing::error!("chainsync error: {:?}", error);
+                            }
+                        }
+                    }
+                    Err(error) => {
+                        tracing::error!("chainsync error: {:?}", error);
+                    }
+                }
+            }
+            keepalive_result = keepalive_handle => {
+                match keepalive_result {
+                    Ok(result) => {
+                        match result {
+                            Ok(_) => {}
+                            Err(error) => {
+                                tracing::error!("keepalive error: {:?}", error);
+                            }
+                        }
+                    }
+                    Err(error) => {
+                        tracing::error!("keepalive error: {:?}", error);
+                    }
+                }
+            }
+        }
+        peer.plexer_handle.abort();
+
+        tracing::info!("waiting 10 seconds before reconnecting...");
+        tokio::time::sleep(tokio::time::Duration::from_secs(10)).await;
+    }
 }

pallas-network/Cargo.toml

@@ -17,6 +17,7 @@ itertools = "0.10.5"
 pallas-codec = { version = "=0.20.0", path = "../pallas-codec" }
 pallas-crypto = { version = "=0.20.0", path = "../pallas-crypto" }
 rand = "0.8.5"
+socket2 = "0.5.5"
 thiserror = "1.0.31"
 tokio = { version = "1", features = ["rt", "net", "io-util", "time", "sync", "macros"] }
 tracing = "0.1.37"

pallas-network/src/multiplexer.rs

@@ -81,9 +81,17 @@ pub enum Bearer {
 }
 
 impl Bearer {
-    pub fn connect_tcp(addr: impl tcp::ToSocketAddrs) -> IOResult<Self> {
-        let stream = tcp::TcpStream::connect(addr)?;
-        stream.set_nodelay(true)?;
+    pub async fn connect_tcp(addr: impl ToSocketAddrs) -> Result<Self, tokio::io::Error> {
+        let stream = TcpStream::connect(addr).await?;
+        // add tcp_keepalive
+        let sock_ref = socket2::SockRef::from(&stream);
+        let mut tcp_keepalive = socket2::TcpKeepalive::new();
+        tcp_keepalive = tcp_keepalive.with_time(tokio::time::Duration::from_secs(20));
+        tcp_keepalive = tcp_keepalive.with_interval(tokio::time::Duration::from_secs(20));
+        let _ = sock_ref.set_tcp_keepalive(&tcp_keepalive);
+        // add tcp_nodelay
+        let _ = sock_ref.set_nodelay(true);
+
         Ok(Self::Tcp(stream))
     }