feat: Implement N2C source

src/bin/scrolls/daemon.rs

@@ -16,6 +16,9 @@ trait FromConfig<T> {
 #[serde(tag = "type")]
 pub enum SourceConfig {
     N2N(sources::n2n::Config),
+
+    #[cfg(target_family = "unix")]
+    N2C(sources::n2c::Config),
 }
 
 impl FromConfig<SourceConfig> for sources::Plugin {
@@ -26,6 +29,7 @@ impl FromConfig<SourceConfig> for sources::Plugin {
     ) -> Self {
         match other {
             SourceConfig::N2N(c) => sources::IntoPlugin::plugin(c, chain, intersect),
+            SourceConfig::N2C(c) => sources::IntoPlugin::plugin(c, chain, intersect),
         }
     }
 }

src/lib.rs

@@ -18,8 +18,14 @@ pub enum Error {
     #[error("ouroboros error: {0}")]
     OuroborosError(String),
 
+    #[error("ledger error: {0}")]
+    LedgerError(String),
+
     #[error("{0}")]
     Message(String),
+
+    #[error("{0}")]
+    Custom(String),
 }
 
 impl Error {
@@ -34,4 +40,14 @@ impl Error {
     pub fn ouroboros(error: Box<dyn std::error::Error>) -> Error {
         Error::OuroborosError(format!("{}", error))
     }
+
+    pub fn custom(error: Box<dyn std::error::Error>) -> Error {
+        Error::Custom(format!("{}", error))
+    }
+}
+
+impl From<Box<dyn std::error::Error>> for Error {
+    fn from(err: Box<dyn std::error::Error>) -> Self {
+        Error::custom(err)
+    }
 }

src/model.rs

@@ -1,10 +1,12 @@
-use std::sync::Arc;
+use std::{ops::Deref, sync::Arc};
 
 use pallas::{
     ledger::primitives::{alonzo, byron},
     network::miniprotocols::Point,
 };
 
+use crate::Error;
+
 #[derive(Debug)]
 pub enum ChainSyncCommand {
     RollForward(Point),
@@ -51,6 +53,29 @@ pub enum MultiEraBlock {
     Byron(byron::Block),
 }
 
+impl MultiEraBlock {
+    pub fn point(&self) -> Result<Point, Error> {
+        match self {
+            MultiEraBlock::Byron(x) => match x.deref() {
+                byron::Block::EbBlock(x) => {
+                    let hash = x.header.to_hash();
+                    let slot = x.header.to_abs_slot();
+                    Ok(Point::Specific(slot, hash.to_vec()))
+                }
+                byron::Block::MainBlock(x) => {
+                    let hash = x.header.to_hash();
+                    let slot = x.header.consensus_data.0.to_abs_slot();
+                    Ok(Point::Specific(slot, hash.to_vec()))
+                }
+            },
+            MultiEraBlock::AlonzoCompatible(x) => {
+                let hash = alonzo::crypto::hash_block_header(&x.1.header);
+                Ok(Point::Specific(x.1.header.header_body.slot, hash.to_vec()))
+            }
+        }
+    }
+}
+
 pub type Set = String;
 pub type Member = String;
 pub type Key = String;

src/sources/mod.rs

@@ -2,7 +2,11 @@ use gasket::messaging::FanoutPort;
 
 use crate::{bootstrap, crosscut, model};
 
+#[cfg(target_family = "unix")]
+pub mod n2c;
+
 pub mod n2n;
+pub mod utils;
 
 pub trait Pluggable {
     fn borrow_output_port(&mut self) -> &'_ mut FanoutPort<model::ChainSyncCommandEx>;
@@ -11,18 +15,21 @@ pub trait Pluggable {
 
 pub enum Plugin {
     N2N(n2n::Plugin),
+    N2C(n2c::Plugin),
 }
 
 impl Plugin {
     pub fn borrow_output_port(&mut self) -> &'_ mut FanoutPort<model::ChainSyncCommandEx> {
         match self {
             Plugin::N2N(p) => p.borrow_output_port(),
+            Plugin::N2C(p) => p.borrow_output_port(),
         }
     }
 
     pub fn spawn(self, pipeline: &mut bootstrap::Pipeline) {
         match self {
             Plugin::N2N(p) => p.spawn(pipeline),
+            Plugin::N2C(p) => p.spawn(pipeline),
         }
     }
 }

src/sources/n2c/chainsync.rs

@@ -0,0 +1,170 @@
+use std::{collections::HashMap, ops::Deref};
+
+use pallas::network::{
+    miniprotocols::{self, chainsync, Error, Point},
+    multiplexer::Channel,
+};
+
+use gasket::{
+    error::AsWorkError,
+    metrics::{Counter, Gauge},
+};
+
+use crate::{
+    model::{ChainSyncCommandEx, MultiEraBlock},
+    sources::utils,
+};
+
+struct ChainObserver {
+    min_depth: usize,
+    output: gasket::messaging::FanoutPort<ChainSyncCommandEx>,
+    chain_buffer: chainsync::RollbackBuffer,
+    blocks: HashMap<Point, MultiEraBlock>,
+    block_count: gasket::metrics::Counter,
+    chain_tip: Gauge,
+}
+
+impl ChainObserver {
+    fn new(
+        min_depth: usize,
+        block_count: Counter,
+        chain_tip: Gauge,
+        output: gasket::messaging::FanoutPort<ChainSyncCommandEx>,
+    ) -> Self {
+        Self {
+            min_depth,
+            block_count,
+            chain_tip,
+            output,
+            chain_buffer: Default::default(),
+            blocks: Default::default(),
+        }
+    }
+}
+
+impl chainsync::Observer<chainsync::BlockContent> for ChainObserver {
+    fn on_roll_forward(
+        &mut self,
+        content: chainsync::BlockContent,
+        tip: &chainsync::Tip,
+    ) -> Result<chainsync::Continuation, Error> {
+        // parse the block and extract the point of the chain
+        let cbor = Vec::from(content.deref());
+        let block = utils::parse_block_content(&cbor)?;
+        let point = block.point()?;
+
+        // store the block for later retrieval
+        self.blocks.insert(point.clone(), block);
+
+        // track the new point in our memory buffer
+        log::info!("rolling forward to point {:?}", point);
+        self.chain_buffer.roll_forward(point);
+
+        // see if we have points that already reached certain depth
+        let ready = self.chain_buffer.pop_with_depth(self.min_depth);
+        log::debug!("found {} points with required min depth", ready.len());
+
+        // find confirmed block in memory and send down the pipeline
+        for point in ready {
+            let block = self
+                .blocks
+                .remove(&point)
+                .expect("required block not found in memory");
+
+            self.output.send(ChainSyncCommandEx::roll_forward(block))?;
+            self.block_count.inc(1);
+        }
+
+        // notify chain tip to the pipeline metrics
+        self.chain_tip.set(tip.1 as i64);
+
+        Ok(chainsync::Continuation::Proceed)
+    }
+
+    fn on_rollback(&mut self, point: &Point) -> Result<chainsync::Continuation, Error> {
+        log::info!("rolling block to point {:?}", point);
+
+        match self.chain_buffer.roll_back(point) {
+            chainsync::RollbackEffect::Handled => {
+                log::debug!("handled rollback within buffer {:?}", point);
+            }
+            chainsync::RollbackEffect::OutOfScope => {
+                log::debug!("rollback out of buffer scope, sending event down the pipeline");
+                self.output
+                    .send(ChainSyncCommandEx::roll_back(point.clone()))?;
+            }
+        }
+
+        Ok(chainsync::Continuation::Proceed)
+    }
+}
+
+type OutputPort = gasket::messaging::FanoutPort<ChainSyncCommandEx>;
+type Runner = miniprotocols::Runner<chainsync::BlockConsumer<ChainObserver>>;
+
+pub struct Worker {
+    channel: Channel,
+    pub min_depth: usize,
+    pub known_points: Option<Vec<Point>>,
+    //finalize_config: Option<FinalizeConfig>,
+    runner: Runner,
+    block_count: gasket::metrics::Counter,
+    chain_tip: Gauge,
+}
+
+impl Worker {
+    pub fn new(
+        channel: Channel,
+        min_depth: usize,
+        known_points: Option<Vec<Point>>,
+        output: OutputPort,
+    ) -> Self {
+        let block_count = Counter::default();
+        let chain_tip = Gauge::default();
+
+        let runner = Runner::new(chainsync::Consumer::initial(
+            known_points.clone(),
+            ChainObserver::new(
+                min_depth as usize,
+                block_count.clone(),
+                chain_tip.clone(),
+                output,
+            ),
+        ));
+
+        Self {
+            channel,
+            min_depth,
+            known_points,
+            runner,
+            block_count,
+            chain_tip,
+        }
+    }
+}
+
+impl gasket::runtime::Worker for Worker {
+    fn metrics(&self) -> gasket::metrics::Registry {
+        gasket::metrics::Builder::new()
+            .with_counter("block_count", &self.block_count)
+            .with_gauge("chain_tip", &self.chain_tip)
+            .build()
+    }
+
+    fn bootstrap(&mut self) -> Result<(), gasket::error::Error> {
+        self.runner.start().or_work_err()?;
+
+        Ok(())
+    }
+
+    fn work(&mut self) -> gasket::runtime::WorkResult {
+        match self.runner.run_step(&mut self.channel) {
+            Ok(true) => Ok(gasket::runtime::WorkOutcome::Done),
+            Ok(false) => Ok(gasket::runtime::WorkOutcome::Partial),
+            Err(err) => Err(gasket::error::Error::WorkError(format!(
+                "chainsync agent error {:?}",
+                err
+            ))),
+        }
+    }
+}

src/sources/n2c/mod.rs

@@ -0,0 +1,84 @@
+pub mod chainsync;
+mod transport;
+
+use std::time::Duration;
+
+use gasket::{error::AsWorkError, messaging::FanoutPort, retries};
+
+use serde::Deserialize;
+
+use crate::{bootstrap::Pipeline, crosscut, model::ChainSyncCommandEx};
+
+use self::transport::Transport;
+
+use super::utils;
+
+#[derive(Deserialize)]
+pub struct Config {
+    pub path: String,
+}
+
+pub struct Plugin {
+    config: Config,
+    intersect: crosscut::IntersectConfig,
+    chain: crosscut::ChainWellKnownInfo,
+    output: FanoutPort<ChainSyncCommandEx>,
+}
+
+impl Plugin {
+    fn bootstrap_transport(&self) -> Result<Transport, gasket::error::Error> {
+        gasket::retries::retry_operation(
+            || Transport::setup(&self.config.path, self.chain.magic).or_work_err(),
+            &retries::Policy {
+                max_retries: 5,
+                backoff_factor: 2,
+                backoff_unit: Duration::from_secs(1),
+                max_backoff: Duration::from_secs(60),
+            },
+            None,
+        )
+    }
+}
+
+impl super::Pluggable for Plugin {
+    fn borrow_output_port(&mut self) -> &'_ mut FanoutPort<ChainSyncCommandEx> {
+        &mut self.output
+    }
+
+    fn spawn(self, pipeline: &mut Pipeline) {
+        let mut transport = self
+            .bootstrap_transport()
+            .expect("transport should be connected after several retries");
+
+        let mut cs_channel = transport.muxer.use_channel(5);
+
+        let known_points =
+            utils::define_known_points(&self.chain, &self.intersect, &mut cs_channel)
+                .expect("chainsync known-points should be defined");
+
+        pipeline.register_stage(
+            "n2c",
+            gasket::runtime::spawn_stage(
+                self::chainsync::Worker::new(cs_channel, 0, known_points, self.output),
+                gasket::runtime::Policy::default(),
+            ),
+        );
+    }
+}
+
+impl super::IntoPlugin for Config {
+    fn plugin(
+        self,
+        chain: &crosscut::ChainWellKnownInfo,
+        intersect: &crosscut::IntersectConfig,
+    ) -> super::Plugin {
+        let plugin = Plugin {
+            config: self,
+            intersect: intersect.clone(),
+            chain: chain.clone(),
+            output: Default::default(),
+        };
+
+        super::Plugin::N2C(plugin)
+    }
+}