chore(miniprotocols): Add chain-sync tip test (#199)

pallas-miniprotocols/src/chainsync/client.rs

@@ -158,7 +158,7 @@ where
         Ok(())
     }
 
-    pub fn recv_request_response(&mut self) -> Result<NextResponse<O>, Error> {
+    pub fn recv_while_can_await(&mut self) -> Result<NextResponse<O>, Error> {
         match self.recv_message()? {
             Message::AwaitReply => {
                 self.0 = State::MustReply;
@@ -176,9 +176,23 @@ where
         }
     }
 
+    pub fn recv_while_must_reply(&mut self) -> Result<NextResponse<O>, Error> {
+        match self.recv_message()? {
+            Message::RollForward(a, b) => {
+                self.0 = State::Idle;
+                Ok(NextResponse::RollForward(a, b))
+            }
+            Message::RollBackward(a, b) => {
+                self.0 = State::Idle;
+                Ok(NextResponse::RollBackward(a, b))
+            }
+            _ => Err(Error::InvalidInbound),
+        }
+    }
+
     pub fn request_next(&mut self) -> Result<NextResponse<O>, Error> {
         self.send_request_next()?;
-        self.recv_request_response()
+        self.recv_while_can_await()
     }
 
     pub fn intersect_origin(&mut self) -> Result<Point, Error> {

pallas-miniprotocols/tests/integration.rs

@@ -4,16 +4,20 @@ use pallas_miniprotocols::{
     handshake::{self, Confirmation},
     Point,
 };
-use pallas_multiplexer::{bearers::Bearer, StdPlexer};
+use pallas_multiplexer::{bearers::Bearer, StdChannel, StdPlexer};
 
-#[test]
+struct N2NChannels {
-#[ignore]
+    channel2: StdChannel,
-pub fn chainsync_happy_path() {
+    channel3: StdChannel,
+}
+
+fn setup_n2n_connection() -> N2NChannels {
     let bearer = Bearer::connect_tcp("preview-node.world.dev.cardano.org:30002").unwrap();
     let mut plexer = StdPlexer::new(bearer);
 
     let channel0 = plexer.use_channel(0);
     let channel2 = plexer.use_channel(2);
+    let channel3 = plexer.use_channel(3);
 
     plexer.muxer.spawn();
     plexer.demuxer.spawn();
@@ -30,6 +34,14 @@ pub fn chainsync_happy_path() {
         assert!(v >= 7);
     }
 
+    N2NChannels { channel2, channel3 }
+}
+
+#[test]
+#[ignore]
+pub fn chainsync_history_happy_path() {
+    let N2NChannels { channel2, .. } = setup_n2n_connection();
+
     let known_point = Point::Specific(
         5953863,
         hex::decode("7e44cb1e230b686875ae6a256b95c9b4eea7c9e9a9d046b626ed69d4c1b9bfe1").unwrap(),
@@ -73,28 +85,48 @@ pub fn chainsync_happy_path() {
 
 #[test]
 #[ignore]
-pub fn blockfetch_happy_path() {
+pub fn chainsync_tip_happy_path() {
-    let bearer = Bearer::connect_tcp("preview-node.world.dev.cardano.org:30002").unwrap();
+    let N2NChannels { channel2, .. } = setup_n2n_connection();
-    let mut plexer = StdPlexer::new(bearer);
 
-    let channel0 = plexer.use_channel(0);
+    let mut client = chainsync::N2NClient::new(channel2);
-    let channel3 = plexer.use_channel(3);
 
-    plexer.muxer.spawn();
+    client.intersect_tip().unwrap();
-    plexer.demuxer.spawn();
 
-    let mut client = handshake::N2NClient::new(channel0);
+    assert!(matches!(client.state(), chainsync::State::Idle));
 
-    let confirmation = client
+    let next = client.request_next().unwrap();
-        .handshake(handshake::n2n::VersionTable::v7_and_above(2))
-        .unwrap();
 
-    assert!(matches!(confirmation, Confirmation::Accepted(..)));
+    assert!(matches!(next, NextResponse::RollBackward(..)));
 
-    if let Confirmation::Accepted(v, _) = confirmation {
+    let mut await_count = 0;
-        assert!(v >= 7);
+
+    for _ in 0..4 {
+        let next = if client.has_agency() {
+            client.request_next().unwrap()
+        } else {
+            await_count += 1;
+            client.recv_while_must_reply().unwrap()
+        };
+
+        match next {
+            NextResponse::RollForward(_, _) => (),
+            NextResponse::Await => (),
+            _ => panic!("expected roll-forward or await"),
+        }
     }
 
+    assert!(await_count > 0, "tip was never reached");
+
+    client.send_done().unwrap();
+
+    assert!(matches!(client.state(), chainsync::State::Done));
+}
+
+#[test]
+#[ignore]
+pub fn blockfetch_happy_path() {
+    let N2NChannels { channel3, .. } = setup_n2n_connection();
+
     let known_point = Point::Specific(
         5953863,
         hex::decode("7e44cb1e230b686875ae6a256b95c9b4eea7c9e9a9d046b626ed69d4c1b9bfe1").unwrap(),