chore: Upgrade to Pallas 0.5.0-alpha.1 (#148)

.gitignore

@@ -9,5 +9,6 @@ Cargo.lock
 # These are backup files generated by rustfmt
 **/*.rs.bk
 
-# Local testdrive environment, mainly for development
-testdrive/custom
+# Local testdrive environments, mainly for development
+testdrive/custom
+testdrive/assert

Cargo.toml

@@ -12,7 +12,7 @@ authors = ["Santiago Carmuega <santiago@carmuega.me>"]
 
 
 [dependencies]
-pallas = "0.5.0-alpha"
+pallas = "0.5.0-alpha.1"
 # pallas = { path = "../pallas/pallas" }
 hex = "0.4.3"
 net2 = "0.2.37"

src/sources/n2c/blocks.rs

@@ -1,39 +1,32 @@
 use std::ops::Deref;
 
 use pallas::{
-    ledger::primitives::{alonzo, byron, probing, Fragment},
-    network::miniprotocols::{
-        chainsync, DecodePayload, EncodePayload, PayloadDecoder, PayloadEncoder, Point,
-    },
+    ledger::primitives::{alonzo, byron, probing},
+    network::miniprotocols::{chainsync::BlockContent, Point},
 };
 
 use crate::Error;
 
 #[derive(Debug)]
 pub(crate) enum MultiEraBlock {
-    Byron(Box<byron::Block>, Vec<u8>),
-    Shelley(Box<alonzo::Block>, Vec<u8>),
+    Byron(Box<byron::Block>),
+    Shelley(Box<alonzo::Block>),
 }
 
-impl EncodePayload for MultiEraBlock {
-    fn encode_payload(&self, _e: &mut PayloadEncoder) -> Result<(), Box<dyn std::error::Error>> {
-        todo!()
-    }
-}
+impl TryFrom<BlockContent> for MultiEraBlock {
+    type Error = Error;
 
-impl DecodePayload for MultiEraBlock {
-    fn decode_payload(d: &mut PayloadDecoder) -> Result<Self, Box<dyn std::error::Error>> {
-        d.tag()?;
-        let bytes = d.bytes()?;
+    fn try_from(value: BlockContent) -> Result<Self, Self::Error> {
+        let bytes = value.deref();
 
         match probing::probe_block_cbor(bytes) {
             probing::BlockInference::Byron => {
-                let block = byron::Block::decode_fragment(bytes)?;
-                Ok(MultiEraBlock::Byron(Box::new(block), Vec::from(bytes)))
+                let block = minicbor::decode(bytes)?;
+                Ok(MultiEraBlock::Byron(Box::new(block)))
             }
             probing::BlockInference::Shelley => {
-                let alonzo::BlockWrapper(_, block) = alonzo::BlockWrapper::decode_fragment(bytes)?;
-                Ok(MultiEraBlock::Shelley(Box::new(block), Vec::from(bytes)))
+                let alonzo::BlockWrapper(_, block) = minicbor::decode(bytes)?;
+                Ok(MultiEraBlock::Shelley(Box::new(block)))
             }
             probing::BlockInference::Inconclusive => {
                 log::error!("CBOR hex for debubbing: {}", hex::encode(bytes));
@@ -43,10 +36,11 @@ impl DecodePayload for MultiEraBlock {
     }
 }
 
-impl chainsync::BlockLike for MultiEraBlock {
-    fn block_point(&self) -> Result<Point, Error> {
+#[allow(unused)]
+impl MultiEraBlock {
+    fn read_cursor(&self) -> Result<Point, Error> {
         match self {
-            MultiEraBlock::Byron(x, _) => match x.deref() {
+            MultiEraBlock::Byron(x) => match x.deref() {
                 byron::Block::EbBlock(x) => {
                     let hash = x.header.to_hash();
                     let slot = x.header.to_abs_slot();
@@ -58,7 +52,7 @@ impl chainsync::BlockLike for MultiEraBlock {
                     Ok(Point(slot, hash.to_vec()))
                 }
             },
-            MultiEraBlock::Shelley(x, _) => {
+            MultiEraBlock::Shelley(x) => {
                 let hash = alonzo::crypto::hash_block_header(&x.header);
                 Ok(Point(x.header.header_body.slot, hash.to_vec()))
             }

src/sources/n2c/run.rs

@@ -1,12 +1,9 @@
-use std::fmt::Debug;
+use std::{fmt::Debug, ops::Deref};
 
 use log::{error, info};
 
 use pallas::network::{
-    miniprotocols::{
-        chainsync::{Consumer, Observer, Tip},
-        run_agent, Point,
-    },
+    miniprotocols::{chainsync, run_agent, Point},
     multiplexer::Channel,
 };
 
@@ -23,23 +20,32 @@ impl Debug for ChainObserver {
     }
 }
 
+impl Deref for ChainObserver {
+    type Target = EventWriter;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
 impl ChainObserver {
     fn new(writer: EventWriter) -> Self {
         Self(writer)
     }
 }
 
-impl Observer<MultiEraBlock> for ChainObserver {
-    fn on_block(
+impl chainsync::Observer<chainsync::BlockContent> for ChainObserver {
+    fn on_roll_forward(
         &self,
-        _cursor: &Option<Point>,
-        content: &MultiEraBlock,
+        content: chainsync::BlockContent,
+        _tip: &chainsync::Tip,
     ) -> Result<(), Box<dyn std::error::Error>> {
-        let Self(writer) = self;
+        let cbor = Vec::from(content.deref());
+        let block = MultiEraBlock::try_from(content)?;
 
-        match content {
-            MultiEraBlock::Byron(model, cbor) => writer.crawl_byron_with_cbor(model, cbor)?,
-            MultiEraBlock::Shelley(model, cbor) => writer.crawl_shelley_with_cbor(model, cbor)?,
+        match block {
+            MultiEraBlock::Byron(model) => self.crawl_byron_with_cbor(&model, &cbor)?,
+            MultiEraBlock::Shelley(model) => self.crawl_shelley_with_cbor(&model, &cbor)?,
         };
 
         Ok(())
@@ -52,7 +58,7 @@ impl Observer<MultiEraBlock> for ChainObserver {
         })
     }
 
-    fn on_intersect_found(&self, point: &Point, _tip: &Tip) -> Result<(), Error> {
+    fn on_intersect_found(&self, point: &Point, _tip: &chainsync::Tip) -> Result<(), Error> {
         info!("intersect found {:?}", point);
         Ok(())
     }
@@ -69,7 +75,7 @@ pub(crate) fn observe_forever(
     from: Point,
 ) -> Result<(), Error> {
     let observer = ChainObserver::new(writer);
-    let agent = Consumer::<MultiEraBlock, _>::initial(vec![from], observer);
+    let agent = chainsync::BlockConsumer::initial(vec![from], observer);
     let agent = run_agent(agent, &mut channel)?;
     error!("chainsync agent final state: {:?}", agent.state);
 

src/sources/n2n/headers.rs

@@ -1,8 +1,6 @@
 use pallas::{
-    ledger::primitives::{alonzo, byron, Fragment},
-    network::miniprotocols::{
-        chainsync, DecodePayload, EncodePayload, PayloadDecoder, PayloadEncoder, Point,
-    },
+    ledger::primitives::{alonzo, byron},
+    network::miniprotocols::{chainsync::HeaderContent, Point},
 };
 
 use crate::Error;
@@ -14,54 +12,31 @@ pub enum MultiEraHeader {
     Shelley(alonzo::Header),
 }
 
-impl EncodePayload for MultiEraHeader {
-    fn encode_payload(&self, _e: &mut PayloadEncoder) -> Result<(), Error> {
-        todo!()
-    }
-}
-
-impl DecodePayload for MultiEraHeader {
-    fn decode_payload(d: &mut PayloadDecoder) -> Result<Self, Error> {
-        d.array()?;
-        let variant = d.u32()?; // WTF is this value?
-
-        match variant {
-            // byron
-            0 => {
-                d.array()?;
-
-                // can't find a reference anywhere about the structure of these values, but they
-                // seem to provide the Byron-specific variant of the header
-                let (block_type, _): (u8, u64) = d.decode()?;
+impl TryFrom<HeaderContent> for MultiEraHeader {
+    type Error = Error;
 
-                d.tag()?;
-                let bytes = d.bytes()?;
-
-                match block_type {
-                    0 => {
-                        let header = byron::EbbHead::decode_fragment(bytes).unwrap();
-                        Ok(MultiEraHeader::ByronBoundary(header))
-                    }
-                    _ => {
-                        let header = byron::BlockHead::decode_fragment(bytes).unwrap();
-                        Ok(MultiEraHeader::Byron(header))
-                    }
+    fn try_from(value: HeaderContent) -> Result<Self, Self::Error> {
+        match value {
+            HeaderContent::Byron(variant, _, bytes) => match variant {
+                0 => {
+                    let header = minicbor::decode(&bytes)?;
+                    Ok(MultiEraHeader::ByronBoundary(header))
                 }
-            }
-            // shelley
-            _ => {
-                d.tag()?;
-                let bytes = d.bytes()?;
-                let header = alonzo::Header::decode_fragment(bytes)?;
-
+                _ => {
+                    let header = minicbor::decode(&bytes)?;
+                    Ok(MultiEraHeader::Byron(header))
+                }
+            },
+            HeaderContent::Shelley(bytes) => {
+                let header = minicbor::decode(&bytes)?;
                 Ok(MultiEraHeader::Shelley(header))
             }
         }
     }
 }
 
-impl chainsync::BlockLike for MultiEraHeader {
-    fn block_point(&self) -> Result<Point, Error> {
+impl MultiEraHeader {
+    pub fn read_cursor(&self) -> Result<Point, Error> {
         match self {
             MultiEraHeader::ByronBoundary(x) => {
                 let hash = x.to_hash();

src/sources/n2n/run.rs

@@ -5,7 +5,11 @@ use log::{info, warn};
 use pallas::{
     ledger::primitives::probing,
     network::{
-        miniprotocols::{blockfetch, chainsync, run_agent, Point},
+        miniprotocols::{
+            blockfetch,
+            chainsync::{self, HeaderContent},
+            run_agent, Point,
+        },
         multiplexer::Channel,
     },
 };
@@ -60,13 +64,17 @@ impl Debug for ChainObserver {
     }
 }
 
-impl chainsync::Observer<MultiEraHeader> for ChainObserver {
-    fn on_block(&self, cursor: &Option<Point>, _content: &MultiEraHeader) -> Result<(), Error> {
-        info!("requesting block fetch for point {:?}", cursor);
+impl chainsync::Observer<HeaderContent> for ChainObserver {
+    fn on_roll_forward(
+        &self,
+        content: chainsync::HeaderContent,
+        _tip: &chainsync::Tip,
+    ) -> Result<(), Error> {
+        let header = MultiEraHeader::try_from(content)?;
+        let cursor = header.read_cursor()?;
 
-        if let Some(cursor) = cursor {
-            self.block_requests.send(cursor.clone())?;
-        }
+        info!("requesting block fetch for point {:?}", cursor);
+        self.block_requests.send(cursor)?;
 
         Ok(())
     }
@@ -105,7 +113,7 @@ pub(crate) fn observe_headers_forever(
         block_requests,
     };
 
-    let agent = chainsync::Consumer::<MultiEraHeader, _>::initial(vec![from], observer);
+    let agent = chainsync::HeaderConsumer::initial(vec![from], observer);
     let agent = run_agent(agent, &mut channel)?;
     warn!("chainsync agent final state: {:?}", agent.state);