diff --git a/src/bin/scrolls/daemon.rs b/src/bin/scrolls/daemon.rs index fed3776e..afaef9c3 100644 --- a/src/bin/scrolls/daemon.rs +++ b/src/bin/scrolls/daemon.rs @@ -16,6 +16,9 @@ trait FromConfig { #[serde(tag = "type")] pub enum SourceConfig { N2N(sources::n2n::Config), + + #[cfg(target_family = "unix")] + N2C(sources::n2c::Config), } impl FromConfig for sources::Plugin { @@ -26,6 +29,7 @@ impl FromConfig for sources::Plugin { ) -> Self { match other { SourceConfig::N2N(c) => sources::IntoPlugin::plugin(c, chain, intersect), + SourceConfig::N2C(c) => sources::IntoPlugin::plugin(c, chain, intersect), } } } diff --git a/src/lib.rs b/src/lib.rs index 69469e11..72118efd 100644 --- a/src/lib.rs +++ b/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) -> Error { Error::OuroborosError(format!("{}", error)) } + + pub fn custom(error: Box) -> Error { + Error::Custom(format!("{}", error)) + } +} + +impl From> for Error { + fn from(err: Box) -> Self { + Error::custom(err) + } } diff --git a/src/model.rs b/src/model.rs index c1c29b5f..8a3fdc6e 100644 --- a/src/model.rs +++ b/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 { + 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; diff --git a/src/sources/mod.rs b/src/sources/mod.rs index 19327a5b..8f42995b 100644 --- a/src/sources/mod.rs +++ b/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; @@ -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 { 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), } } } diff --git a/src/sources/n2c/chainsync.rs b/src/sources/n2c/chainsync.rs new file mode 100644 index 00000000..1e58d27d --- /dev/null +++ b/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, + chain_buffer: chainsync::RollbackBuffer, + blocks: HashMap, + 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, + ) -> Self { + Self { + min_depth, + block_count, + chain_tip, + output, + chain_buffer: Default::default(), + blocks: Default::default(), + } + } +} + +impl chainsync::Observer for ChainObserver { + fn on_roll_forward( + &mut self, + content: chainsync::BlockContent, + tip: &chainsync::Tip, + ) -> Result { + // 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 { + 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; +type Runner = miniprotocols::Runner>; + +pub struct Worker { + channel: Channel, + pub min_depth: usize, + pub known_points: Option>, + //finalize_config: Option, + runner: Runner, + block_count: gasket::metrics::Counter, + chain_tip: Gauge, +} + +impl Worker { + pub fn new( + channel: Channel, + min_depth: usize, + known_points: Option>, + 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 + ))), + } + } +} diff --git a/src/sources/n2c/mod.rs b/src/sources/n2c/mod.rs new file mode 100644 index 00000000..142f4dcb --- /dev/null +++ b/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, +} + +impl Plugin { + fn bootstrap_transport(&self) -> Result { + 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 { + &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) + } +} diff --git a/src/sources/n2c/transport.rs b/src/sources/n2c/transport.rs new file mode 100644 index 00000000..db15075b --- /dev/null +++ b/src/sources/n2c/transport.rs @@ -0,0 +1,46 @@ +use std::os::unix::net::UnixStream; + +use pallas::network::{ + miniprotocols::{self, handshake}, + multiplexer::Multiplexer, +}; + +pub struct Transport { + pub muxer: Multiplexer, + pub version: handshake::VersionNumber, +} + +impl Transport { + fn connect_muxer(address: &str) -> Result { + log::debug!("connecting muxer"); + let unix = UnixStream::connect(address)?; + let muxer = Multiplexer::setup(unix, &[0, 5])?; + + Ok(muxer) + } + + fn do_handshake( + muxer: &mut Multiplexer, + magic: u64, + ) -> Result { + log::debug!("doing handshake"); + + let mut channel = muxer.use_channel(0); + let versions = handshake::n2c::VersionTable::v1_and_above(magic); + let agent = + miniprotocols::run_agent(handshake::Initiator::initial(versions), &mut channel)?; + log::info!("handshake output: {:?}", agent.output); + + match agent.output { + handshake::Output::Accepted(version, _) => Ok(version), + _ => Err("couldn't agree on handshake version".into()), + } + } + + pub fn setup(address: &str, magic: u64) -> Result { + let mut muxer = Self::connect_muxer(address)?; + let version = Self::do_handshake(&mut muxer, magic)?; + + Ok(Self { muxer, version }) + } +} diff --git a/src/sources/n2n/blockfetch.rs b/src/sources/n2n/blockfetch.rs index 06b89fde..048450b9 100644 --- a/src/sources/n2n/blockfetch.rs +++ b/src/sources/n2n/blockfetch.rs @@ -1,14 +1,13 @@ -use pallas::{ - ledger::primitives::{alonzo, byron, probing, Era, Fragment}, - network::{ - miniprotocols::{blockfetch, run_agent, Error, Point}, - multiplexer::Channel, - }, +use pallas::network::{ + miniprotocols::{blockfetch, run_agent, Error, Point}, + multiplexer::Channel, }; use gasket::{error::*, runtime::WorkOutcome}; -use crate::model::{ChainSyncCommand, ChainSyncCommandEx, MultiEraBlock}; +use crate::model::{ChainSyncCommand, ChainSyncCommandEx}; + +use crate::sources::utils; struct Observer<'a> { output: &'a mut FanoutPort, @@ -16,22 +15,7 @@ struct Observer<'a> { impl<'a> blockfetch::Observer for Observer<'a> { fn on_block_received(&mut self, body: Vec) -> Result<(), Error> { - let block = match probing::probe_block_cbor_era(&body) { - probing::Outcome::Matched(era) => match era { - Era::Byron => MultiEraBlock::Byron(byron::Block::decode_fragment(&body)?), - _ => MultiEraBlock::AlonzoCompatible(alonzo::BlockWrapper::decode_fragment(&body)?), - }, - // TODO: we're assuming that the genesis block is Byron-compatible. Is this a safe - // assumption? - probing::Outcome::GenesisBlock => { - MultiEraBlock::Byron(byron::Block::decode_fragment(&body)?) - } - probing::Outcome::Inconclusive => { - let msg = format!("can't infer primitive block from cbor, inconclusive probing. CBOR hex for debugging: {}", hex::encode(body)); - return Err(msg.into()); - } - }; - + let block = utils::parse_block_content(&body)?; self.output.send(ChainSyncCommandEx::roll_forward(block))?; Ok(()) diff --git a/src/sources/n2n/mod.rs b/src/sources/n2n/mod.rs index c695ede3..4d0bcf0b 100644 --- a/src/sources/n2n/mod.rs +++ b/src/sources/n2n/mod.rs @@ -11,10 +11,7 @@ use gasket::{ retries, }; pub use messages::*; -use pallas::network::{ - miniprotocols::{chainsync::TipFinder, run_agent, Point}, - multiplexer::Channel, -}; + use serde::Deserialize; use crate::{ @@ -25,6 +22,8 @@ use crate::{ use self::transport::Transport; +use super::utils; + #[derive(Deserialize)] pub struct Config { pub address: String, @@ -50,43 +49,6 @@ impl Plugin { None, ) } - - fn find_end_of_chain(&self, channel: &mut Channel) -> Result { - let point = Point::Specific( - self.chain.shelley_known_slot, - hex::decode(&self.chain.shelley_known_hash) - .map_err(|_| crate::Error::config("can't decode shelley known hash"))?, - ); - - let agent = TipFinder::initial(point); - let agent = run_agent(agent, channel).map_err(crate::Error::ouroboros)?; - - match agent.output { - Some(tip) => Ok(tip.0), - None => Err(crate::Error::message("failure acquiring end of chain")), - } - } - - fn define_known_points( - &self, - channel: &mut Channel, - ) -> Result>, crate::Error> { - match &self.intersect { - crosscut::IntersectConfig::Origin => Ok(None), - crosscut::IntersectConfig::Tip => { - let tip = self.find_end_of_chain(channel)?; - Ok(Some(vec![tip])) - } - crosscut::IntersectConfig::Point(x) => { - let point = x.clone().try_into()?; - Ok(Some(vec![point])) - } - crosscut::IntersectConfig::Fallbacks(x) => { - let points: Result, _> = x.iter().cloned().map(|x| x.try_into()).collect(); - Ok(Some(points?)) - } - } - } } impl super::Pluggable for Plugin { @@ -102,9 +64,9 @@ impl super::Pluggable for Plugin { let mut cs_channel = transport.muxer.use_channel(2); let bf_channel = transport.muxer.use_channel(3); - let known_points = self - .define_known_points(&mut cs_channel) - .expect("chainsync known-points should be defined"); + let known_points = + utils::define_known_points(&self.chain, &self.intersect, &mut cs_channel) + .expect("chainsync known-points should be defined"); let mut headers_out = OutputPort::::default(); let mut headers_in = InputPort::::default(); diff --git a/src/sources/utils.rs b/src/sources/utils.rs new file mode 100644 index 00000000..0a623b25 --- /dev/null +++ b/src/sources/utils.rs @@ -0,0 +1,82 @@ +use pallas::{ + ledger::primitives::{alonzo, byron, probing, Era, Fragment}, + network::{ + miniprotocols::{chainsync::TipFinder, run_agent, Point}, + multiplexer::Channel, + }, +}; + +use crate::{ + crosscut::{self, ChainWellKnownInfo, IntersectConfig}, + model::MultiEraBlock, + Error, +}; + +pub fn parse_block_content(body: &[u8]) -> Result { + match probing::probe_block_cbor_era(&body) { + probing::Outcome::Matched(era) => match era { + Era::Byron => { + let primitive = byron::Block::decode_fragment(&body)?; + let block = MultiEraBlock::Byron(primitive); + Ok(block) + } + _ => { + let primitive = alonzo::BlockWrapper::decode_fragment(&body)?; + let block = MultiEraBlock::AlonzoCompatible(primitive); + Ok(block) + } + }, + // TODO: we're assuming that the genesis block is Byron-compatible. Is this a safe + // assumption? + probing::Outcome::GenesisBlock => { + let primitive = byron::Block::decode_fragment(&body)?; + let block = MultiEraBlock::Byron(primitive); + Ok(block) + } + probing::Outcome::Inconclusive => { + let msg = format!("can't infer primitive block from cbor, inconclusive probing. CBOR hex for debugging: {}", hex::encode(body)); + return Err(Error::Message(msg)); + } + } +} + +pub fn find_end_of_chain( + chain: &ChainWellKnownInfo, + channel: &mut Channel, +) -> Result { + let point = Point::Specific( + chain.shelley_known_slot, + hex::decode(&chain.shelley_known_hash) + .map_err(|_| crate::Error::config("can't decode shelley known hash"))?, + ); + + let agent = TipFinder::initial(point); + let agent = run_agent(agent, channel).map_err(crate::Error::ouroboros)?; + + match agent.output { + Some(tip) => Ok(tip.0), + None => Err(crate::Error::message("failure acquiring end of chain")), + } +} + +pub fn define_known_points( + chain: &ChainWellKnownInfo, + intersect: &IntersectConfig, + channel: &mut Channel, +) -> Result>, crate::Error> { + match &intersect { + crosscut::IntersectConfig::Origin => Ok(None), + crosscut::IntersectConfig::Tip => { + let tip = find_end_of_chain(chain, channel)?; + Ok(Some(vec![tip])) + } + crosscut::IntersectConfig::Point(x) => { + let point = x.clone().try_into()?; + Ok(Some(vec![point])) + } + crosscut::IntersectConfig::Fallbacks(x) => { + let points: Result, _> = x.iter().cloned().map(|x| x.try_into()).collect(); + Ok(Some(points?)) + } + } +}