feat: Introduce 'intersect' argument

Cargo.toml

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

src/bin/oura/dump.rs

@@ -6,7 +6,7 @@ use clap::ArgMatches;
 use oura::{
     mapper::Config as MapperConfig,
     pipelining::{BootstrapResult, SinkProvider, SourceProvider, StageReceiver},
-    sources::{AddressArg, BearerKind, MagicArg},
+    sources::{AddressArg, BearerKind, IntersectArg, MagicArg},
     utils::{ChainWellKnownInfo, Utils, WithUtils},
 };
 
@@ -77,8 +77,8 @@ pub fn run(args: &ArgMatches) -> Result<(), Error> {
         false => MagicArg::default(),
     };
 
-    let since = match args.is_present("since") {
-        true => Some(args.value_of_t("since")?),
+    let intersect = match args.is_present("since") {
+        true => Some(IntersectArg::Point(args.value_of_t("since")?)),
         false => None,
     };
 
@@ -116,17 +116,17 @@ pub fn run(args: &ArgMatches) -> Result<(), Error> {
             well_known: None,
             min_depth: 0,
             mapper,
-            since,
-            intersections: None,
+            since: None,
+            intersect,
         }),
         PeerMode::AsClient => DumpSource::N2C(N2CConfig {
             address: AddressArg(bearer, socket),
             magic: Some(magic),
             well_known: None,
             min_depth: 0,
             mapper,
-            since,
-            intersections: None,
+            since: None,
+            intersect,
         }),
     };
 

src/bin/oura/watch.rs

@@ -4,7 +4,7 @@ use clap::ArgMatches;
 use oura::{
     mapper::Config as MapperConfig,
     pipelining::{SinkProvider, SourceProvider},
-    sources::{AddressArg, BearerKind, MagicArg},
+    sources::{AddressArg, BearerKind, IntersectArg, MagicArg},
     utils::{ChainWellKnownInfo, Utils, WithUtils},
 };
 
@@ -58,8 +58,8 @@ pub fn run(args: &ArgMatches) -> Result<(), Error> {
         false => MagicArg::default(),
     };
 
-    let since = match args.is_present("since") {
-        true => Some(args.value_of_t("since")?),
+    let intersect = match args.is_present("since") {
+        true => Some(IntersectArg::Point(args.value_of_t("since")?)),
         false => None,
     };
 
@@ -94,17 +94,17 @@ pub fn run(args: &ArgMatches) -> Result<(), Error> {
             well_known: None,
             min_depth: 0,
             mapper,
-            since,
-            intersections: None,
+            since: None,
+            intersect,
         }),
         PeerMode::AsClient => WatchSource::N2C(N2CConfig {
             address: AddressArg(bearer, socket),
             magic: Some(magic),
             well_known: None,
             min_depth: 0,
             mapper,
-            since,
-            intersections: None,
+            since: None,
+            intersect,
         }),
     };
 

src/sources/common.rs

@@ -141,6 +141,15 @@ where
     deserializer.deserialize_any(MagicArgVisitor)
 }
 
+#[derive(Debug, Deserialize)]
+#[serde(tag = "type", content = "value")]
+pub enum IntersectArg {
+    Tip,
+    Origin,
+    Point(PointArg),
+    Fallbacks(Vec<PointArg>),
+}
+
 pub(crate) fn find_end_of_chain(
     channel: &mut Channel,
     well_known: &ChainWellKnownInfo,
@@ -161,30 +170,61 @@ pub(crate) fn find_end_of_chain(
 }
 
 pub(crate) fn define_start_point(
+    intersect: &Option<IntersectArg>,
     since: &Option<PointArg>,
-    intersections: &Option<Vec<PointArg>>,
     utils: &Utils,
     cs_channel: &mut Channel,
-) -> Result<Vec<Point>, Error> {
+) -> Result<Option<Vec<Point>>, Error> {
     let cursor = utils.get_cursor_if_any();
 
-    match (cursor, since, intersections) {
-        (Some(cursor), _, _) => {
+    match cursor {
+        Some(cursor) => {
             log::info!("found persisted cursor, will use as starting point");
-            Ok(vec![cursor.try_into()?])
-        }
-        (None, Some(arg), _) => {
-            log::info!("explicit 'since' argument, will use as starting point");
-            Ok(vec![arg.clone().try_into()?])
-        }
-        (None, None, Some(args)) => {
-            log::info!("intersections argument, will use as starting point");
-            args.iter().map(|x| x.clone().try_into()).collect()
-        }
-        _ => {
-            log::info!("no starting point specified, will use tip of chain");
-            let point = find_end_of_chain(cs_channel, &utils.well_known)?;
-            Ok(vec![point])
+            let points = vec![cursor.try_into()?];
+
+            Ok(Some(points))
         }
+        None => match intersect {
+            Some(IntersectArg::Fallbacks(x)) => {
+                log::info!("found 'fallbacks' intersect argument, will use as starting point");
+                let points: Result<Vec<_>, _> = x.iter().map(|x| x.clone().try_into()).collect();
+
+                Ok(Some(points?))
+            }
+            Some(IntersectArg::Origin) => {
+                log::info!("found 'origin' instersect argument, will use as starting point");
+
+                Ok(None)
+            }
+            Some(IntersectArg::Point(x)) => {
+                log::info!("found 'point' intersect argument, will use as starting point");
+                let points = vec![x.clone().try_into()?];
+
+                Ok(Some(points))
+            }
+            Some(IntersectArg::Tip) => {
+                log::info!("found 'tip' intersect argument, will use as starting point");
+                let tip = find_end_of_chain(cs_channel, &utils.well_known)?;
+                let points = vec![tip];
+
+                Ok(Some(points))
+            }
+            None => match since {
+                Some(x) => {
+                    log::info!("explicit 'since' argument, will use as starting point");
+                    log::warn!("`since` value is deprecated, please use `intersect`");
+                    let points = vec![x.clone().try_into()?];
+
+                    Ok(Some(points))
+                }
+                None => {
+                    log::info!("no starting point specified, will use tip of chain");
+                    let tip = find_end_of_chain(cs_channel, &utils.well_known)?;
+                    let points = vec![tip];
+
+                    Ok(Some(points))
+                }
+            },
+        },
     }
 }

src/sources/n2c/blocks.rs

@@ -30,6 +30,12 @@ impl TryFrom<BlockContent> for MultiEraBlock {
                     Ok(MultiEraBlock::AlonzoCompatible(Box::new(block), era))
                 }
             },
+            // TODO: we're assuming that the genesis block is Byron-compatible. Is this a safe
+            // assumption?
+            probing::Outcome::GenesisBlock => {
+                let block = minicbor::decode(bytes)?;
+                Ok(MultiEraBlock::Byron(Box::new(block)))
+            }
             probing::Outcome::Inconclusive => {
                 log::error!("CBOR hex for debubbing: {}", hex::encode(bytes));
                 Err("can't infer primitive block from cbor, inconslusive probing".into())

src/sources/n2c/run.rs

@@ -109,7 +109,7 @@ impl chainsync::Observer<chainsync::BlockContent> for ChainObserver {
 pub(crate) fn observe_forever(
     mut channel: Channel,
     event_writer: EventWriter,
-    from: Vec<Point>,
+    known_points: Option<Vec<Point>>,
     min_depth: usize,
 ) -> Result<(), Error> {
     let observer = ChainObserver {
@@ -119,7 +119,7 @@ pub(crate) fn observe_forever(
         event_writer,
     };
 
-    let agent = chainsync::BlockConsumer::initial(Some(from), observer);
+    let agent = chainsync::BlockConsumer::initial(known_points, observer);
     let agent = run_agent(agent, &mut channel)?;
     log::warn!("chainsync agent final state: {:?}", agent.state);
 

src/sources/n2c/setup.rs

@@ -7,7 +7,7 @@ use net2::TcpStreamExt;
 use log::info;
 
 use pallas::network::{
-    miniprotocols::{handshake::n2c, run_agent, Point, MAINNET_MAGIC},
+    miniprotocols::{handshake::n2c, run_agent, MAINNET_MAGIC},
     multiplexer::{Channel, Multiplexer},
 };
 
@@ -18,7 +18,7 @@ use crate::{
     pipelining::{new_inter_stage_channel, PartialBootstrapResult, SourceProvider},
     sources::{
         common::{AddressArg, BearerKind, MagicArg, PointArg},
-        define_start_point,
+        define_start_point, IntersectArg,
     },
     utils::{ChainWellKnownInfo, WithUtils},
     Error,
@@ -33,9 +33,10 @@ pub struct Config {
     #[serde(deserialize_with = "crate::sources::common::deserialize_magic_arg")]
     pub magic: Option<MagicArg>,
 
+    #[deprecated(note = "use intersect value instead")]
     pub since: Option<PointArg>,
 
-    pub intersections: Option<Vec<PointArg>>,
+    pub intersect: Option<IntersectArg>,
 
     #[deprecated(note = "chain info is now pipeline-wide, use utils")]
     pub well_known: Option<ChainWellKnownInfo>,
@@ -102,18 +103,20 @@ impl SourceProvider for WithUtils<Config> {
 
         let mut cs_channel = muxer.use_channel(5);
 
-        let since: Vec<Point> = define_start_point(
+        let known_points = define_start_point(
+            &self.inner.intersect,
+            #[allow(deprecated)]
             &self.inner.since,
-            &self.inner.intersections,
             &self.utils,
             &mut cs_channel,
         )?;
 
-        info!("starting from chain point: {:?}", &since);
+        info!("starting chain sync from: {:?}", &known_points);
 
         let min_depth = self.inner.min_depth;
         let handle = std::thread::spawn(move || {
-            observe_forever(cs_channel, writer, since, min_depth).expect("chainsync loop failed");
+            observe_forever(cs_channel, writer, known_points, min_depth)
+                .expect("chainsync loop failed");
         });
 
         Ok((handle, output_rx))

src/sources/n2n/run.rs

@@ -39,6 +39,13 @@ impl blockfetch::Observer for Block2EventMapper {
                         .ok_or_warn("error crawling block for events");
                 }
             },
+            // TODO: we're assuming that the genesis block is Byron-compatible. Is this a safe
+            // assumption?
+            probing::Outcome::GenesisBlock => {
+                writer
+                    .crawl_from_byron_cbor(&body)
+                    .ok_or_warn("error crawling block for events");
+            }
             probing::Outcome::Inconclusive => {
                 log::error!("can't infer primitive block from cbor, inconslusive probing. CBOR hex for debubbing: {}", hex::encode(body));
             }
@@ -138,7 +145,7 @@ pub(crate) fn fetch_blocks_forever(
 pub(crate) fn observe_headers_forever(
     mut channel: Channel,
     event_writer: EventWriter,
-    from: Vec<Point>,
+    known_points: Option<Vec<Point>>,
     block_requests: SyncSender<Point>,
     min_depth: usize,
 ) -> Result<(), Error> {
@@ -149,7 +156,7 @@ pub(crate) fn observe_headers_forever(
         block_requests,
     };
 
-    let agent = chainsync::HeaderConsumer::initial(Some(from), observer);
+    let agent = chainsync::HeaderConsumer::initial(known_points, observer);
     let agent = run_agent(agent, &mut channel)?;
     log::warn!("chainsync agent final state: {:?}", agent.state);