From 1295675874e265026be8856578550b3d7420a8f3 Mon Sep 17 00:00:00 2001 From: Santiago Carmuega Date: Thu, 9 Jun 2022 18:43:35 -0300 Subject: [PATCH] feat: Introduce 'enrich' stage --- src/bootstrap.rs | 4 +- src/model.rs | 5 +- src/reducers/address_by_txo.rs | 3 +- src/reducers/mod.rs | 3 +- src/reducers/total_transactions_count.rs | 2 +- ...ransactions_count_by_contract_addresses.rs | 2 +- .../transactions_count_by_contract_address.rs | 2 +- ...ions_count_by_contract_address_by_epoch.rs | 2 +- src/reducers/transactions_count_by_epoch.rs | 2 +- src/reducers/worker.rs | 2 +- src/sources/mod.rs | 10 +-- src/sources/n2c/chainsync.rs | 74 +++++++++++-------- src/sources/n2c/mod.rs | 25 ++++--- src/sources/n2n/blockfetch.rs | 8 +- src/sources/n2n/chainsync.rs | 26 +++++-- src/sources/n2n/mod.rs | 25 ++++--- src/sources/utils.rs | 2 +- src/storage/mod.rs | 8 +- src/storage/redis.rs | 67 +++++++++++++---- 19 files changed, 179 insertions(+), 93 deletions(-) diff --git a/src/bootstrap.rs b/src/bootstrap.rs index fd583850..5a3f15e4 100644 --- a/src/bootstrap.rs +++ b/src/bootstrap.rs @@ -40,9 +40,11 @@ pub fn build( ); let reader = storage.build_read_plugin(); + source.spawn_stages(&mut pipeline, reader); - source.spawn(&mut pipeline, &reader); + let reader = storage.build_read_plugin(); reducer.spawn_stages(&mut pipeline, reader); + storage.spawn_stages(&mut pipeline); Ok(pipeline) diff --git a/src/model.rs b/src/model.rs index 4f87f32e..438d8069 100644 --- a/src/model.rs +++ b/src/model.rs @@ -80,6 +80,7 @@ pub type Set = String; pub type Member = String; pub type Key = String; pub type Value = String; +pub type Delta = i64; pub type Timestamp = u64; #[derive(Debug)] @@ -90,8 +91,9 @@ pub enum CRDTCommand { TwoPhaseSetRemove(Set, Member), GrowOnlySetAdd(Set, Member), LastWriteWins(Key, Value, Timestamp), + AnyWriteWins(Key, Value), // TODO make sure Value is a generic not stringly typed - PNCounter(Key, Value), + PNCounter(Key, Delta), BlockFinished(Point), } @@ -108,6 +110,7 @@ impl CRDTCommand { } pub enum StateQuery { + KeyValue(Key), LatestKeyValue(Key), SetMembers(Set), } diff --git a/src/reducers/address_by_txo.rs b/src/reducers/address_by_txo.rs index 1642ebfe..9753035a 100644 --- a/src/reducers/address_by_txo.rs +++ b/src/reducers/address_by_txo.rs @@ -3,7 +3,7 @@ use pallas::crypto::hash::Hash; use pallas::ledger::primitives::{alonzo, byron}; use serde::Deserialize; -use crate::{crosscut, model}; +use crate::{crosscut, model, storage}; #[derive(Deserialize)] pub struct Config { @@ -89,6 +89,7 @@ impl Reducer { pub fn reduce_block( &mut self, block: &model::MultiEraBlock, + _state: &mut storage::ReadPlugin, output: &mut super::OutputPort, ) -> Result<(), gasket::error::Error> { match block { diff --git a/src/reducers/mod.rs b/src/reducers/mod.rs index 678712e9..6694c292 100644 --- a/src/reducers/mod.rs +++ b/src/reducers/mod.rs @@ -128,6 +128,7 @@ impl Reducer { pub fn reduce_block( &mut self, block: &model::MultiEraBlock, + state: &mut storage::ReadPlugin, output: &mut OutputPort, ) -> Result<(), gasket::error::Error> { match self { @@ -136,7 +137,7 @@ impl Reducer { Reducer::PoolByStake(x) => x.reduce_block(block, output), #[cfg(feature = "unstable")] - Reducer::AddressByTxo(x) => x.reduce_block(block, output), + Reducer::AddressByTxo(x) => x.reduce_block(block, state, output), #[cfg(feature = "unstable")] Reducer::TotalTransactionsCount(x) => x.reduce_block(block, output), #[cfg(feature = "unstable")] diff --git a/src/reducers/total_transactions_count.rs b/src/reducers/total_transactions_count.rs index 10a0512a..ac2c638f 100644 --- a/src/reducers/total_transactions_count.rs +++ b/src/reducers/total_transactions_count.rs @@ -22,7 +22,7 @@ impl Reducer { None => "total_transactions_count".to_string(), }; - let crdt = model::CRDTCommand::PNCounter(key, 1.to_string()); + let crdt = model::CRDTCommand::PNCounter(key, 1); output.send(gasket::messaging::Message::from(crdt))?; diff --git a/src/reducers/total_transactions_count_by_contract_addresses.rs b/src/reducers/total_transactions_count_by_contract_addresses.rs index 96270d9a..c801eb17 100644 --- a/src/reducers/total_transactions_count_by_contract_addresses.rs +++ b/src/reducers/total_transactions_count_by_contract_addresses.rs @@ -22,7 +22,7 @@ impl Reducer { None => "total_transactions_count_by_contract_addresses".to_string(), }; - let crdt = model::CRDTCommand::PNCounter(key, "1".to_string()); + let crdt = model::CRDTCommand::PNCounter(key, 1); output.send(gasket::messaging::Message::from(crdt))?; Ok(()) diff --git a/src/reducers/transactions_count_by_contract_address.rs b/src/reducers/transactions_count_by_contract_address.rs index 892c1f68..63d298dc 100644 --- a/src/reducers/transactions_count_by_contract_address.rs +++ b/src/reducers/transactions_count_by_contract_address.rs @@ -27,7 +27,7 @@ impl Reducer { None => format!("{}", contract_address.to_string()), }; - let crdt = model::CRDTCommand::PNCounter(key, "1".to_string()); + let crdt = model::CRDTCommand::PNCounter(key, 1); output.send(gasket::messaging::Message::from(crdt))?; } diff --git a/src/reducers/transactions_count_by_contract_address_by_epoch.rs b/src/reducers/transactions_count_by_contract_address_by_epoch.rs index 9569034c..a4ecb043 100644 --- a/src/reducers/transactions_count_by_contract_address_by_epoch.rs +++ b/src/reducers/transactions_count_by_contract_address_by_epoch.rs @@ -39,7 +39,7 @@ impl Reducer { None => format!("{}.{}", contract_address.to_string(), epoch_no), }; - let crdt = model::CRDTCommand::PNCounter(key, "1".to_string()); + let crdt = model::CRDTCommand::PNCounter(key, 1); output.send(gasket::messaging::Message::from(crdt))?; } diff --git a/src/reducers/transactions_count_by_epoch.rs b/src/reducers/transactions_count_by_epoch.rs index f6492c3d..a6f10eb3 100644 --- a/src/reducers/transactions_count_by_epoch.rs +++ b/src/reducers/transactions_count_by_epoch.rs @@ -60,7 +60,7 @@ impl Reducer { let key = format!("{}.{}", prefix, epoch_no.to_string()); - let crdt = model::CRDTCommand::PNCounter(key, 1.to_string()); + let crdt = model::CRDTCommand::PNCounter(key, 1); output.send(gasket::messaging::Message::from(crdt))?; diff --git a/src/reducers/worker.rs b/src/reducers/worker.rs index d54e28e5..8b527d4c 100644 --- a/src/reducers/worker.rs +++ b/src/reducers/worker.rs @@ -37,7 +37,7 @@ impl Worker { ))?; for reducer in self.reducers.iter_mut() { - reducer.reduce_block(block, &mut self.output)?; + reducer.reduce_block(block, &mut self.state, &mut self.output)?; self.ops_count.inc(1); } diff --git a/src/sources/mod.rs b/src/sources/mod.rs index fa9fa825..8d98b83e 100644 --- a/src/sources/mod.rs +++ b/src/sources/mod.rs @@ -1,4 +1,4 @@ -use gasket::messaging::FanoutPort; +use gasket::messaging::OutputPort; use serde::Deserialize; use crate::{bootstrap, crosscut, model, storage}; @@ -37,17 +37,17 @@ pub enum Bootstrapper { } impl Bootstrapper { - pub fn borrow_output_port(&mut self) -> &'_ mut FanoutPort { + pub fn borrow_output_port(&mut self) -> &'_ mut OutputPort { match self { Bootstrapper::N2N(p) => p.borrow_output_port(), Bootstrapper::N2C(p) => p.borrow_output_port(), } } - pub fn spawn(self, pipeline: &mut bootstrap::Pipeline, storage: &storage::ReadPlugin) { + pub fn spawn_stages(self, pipeline: &mut bootstrap::Pipeline, state: storage::ReadPlugin) { match self { - Bootstrapper::N2N(p) => p.spawn(pipeline, storage), - Bootstrapper::N2C(p) => p.spawn_stages(pipeline, storage), + Bootstrapper::N2N(p) => p.spawn_stages(pipeline, state), + Bootstrapper::N2C(p) => p.spawn_stages(pipeline, state), } } } diff --git a/src/sources/n2c/chainsync.rs b/src/sources/n2c/chainsync.rs index 1e58d27d..805cf743 100644 --- a/src/sources/n2c/chainsync.rs +++ b/src/sources/n2c/chainsync.rs @@ -11,13 +11,15 @@ use gasket::{ }; use crate::{ + crosscut, model::{ChainSyncCommandEx, MultiEraBlock}, sources::utils, + storage, }; struct ChainObserver { min_depth: usize, - output: gasket::messaging::FanoutPort, + output: OutputPort, chain_buffer: chainsync::RollbackBuffer, blocks: HashMap, block_count: gasket::metrics::Counter, @@ -25,12 +27,7 @@ struct ChainObserver { } impl ChainObserver { - fn new( - min_depth: usize, - block_count: Counter, - chain_tip: Gauge, - output: gasket::messaging::FanoutPort, - ) -> Self { + fn new(min_depth: usize, block_count: Counter, chain_tip: Gauge, output: OutputPort) -> Self { Self { min_depth, block_count, @@ -99,15 +96,18 @@ impl chainsync::Observer for ChainObserver { } } -type OutputPort = gasket::messaging::FanoutPort; +type OutputPort = gasket::messaging::OutputPort; type Runner = miniprotocols::Runner>; pub struct Worker { channel: Channel, pub min_depth: usize, - pub known_points: Option>, + chain: crosscut::ChainWellKnownInfo, + intersect: crosscut::IntersectConfig, + state: storage::ReadPlugin, + output: OutputPort, //finalize_config: Option, - runner: Runner, + runner: Option, block_count: gasket::metrics::Counter, chain_tip: Gauge, } @@ -116,29 +116,21 @@ impl Worker { pub fn new( channel: Channel, min_depth: usize, - known_points: Option>, + chain: crosscut::ChainWellKnownInfo, + intersect: crosscut::IntersectConfig, + state: storage::ReadPlugin, 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, + chain, + intersect, + state, + output, + runner: None, + block_count: Default::default(), + chain_tip: Default::default(), } } } @@ -152,13 +144,35 @@ impl gasket::runtime::Worker for Worker { } fn bootstrap(&mut self) -> Result<(), gasket::error::Error> { - self.runner.start().or_work_err()?; + self.state.bootstrap().or_work_err()?; + + let known_points = utils::define_known_points( + &self.chain, + &self.intersect, + &mut self.state, + &mut self.channel, + ) + .or_work_err()?; + + let mut runner = Runner::new(chainsync::Consumer::initial( + known_points, + ChainObserver::new( + self.min_depth, + self.block_count.clone(), + self.chain_tip.clone(), + self.output.clone(), + ), + )); + + runner.start().or_work_err()?; + + self.runner = Some(runner); Ok(()) } fn work(&mut self) -> gasket::runtime::WorkResult { - match self.runner.run_step(&mut self.channel) { + match self.runner.as_mut().unwrap().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!( diff --git a/src/sources/n2c/mod.rs b/src/sources/n2c/mod.rs index 6ef80bb0..63381cb6 100644 --- a/src/sources/n2c/mod.rs +++ b/src/sources/n2c/mod.rs @@ -3,7 +3,7 @@ mod transport; use std::time::Duration; -use gasket::{error::AsWorkError, messaging::FanoutPort, retries}; +use gasket::{error::AsWorkError, messaging::OutputPort, retries}; use serde::Deserialize; @@ -11,8 +11,6 @@ use crate::{bootstrap::Pipeline, crosscut, model::ChainSyncCommandEx, storage}; use self::transport::Transport; -use super::utils; - #[derive(Deserialize)] pub struct Config { pub path: String, @@ -37,7 +35,7 @@ pub struct Bootstrapper { config: Config, intersect: crosscut::IntersectConfig, chain: crosscut::ChainWellKnownInfo, - output: FanoutPort, + output: OutputPort, } impl Bootstrapper { @@ -54,25 +52,28 @@ impl Bootstrapper { ) } - pub fn borrow_output_port(&mut self) -> &'_ mut FanoutPort { + pub fn borrow_output_port(&mut self) -> &'_ mut OutputPort { &mut self.output } - pub fn spawn_stages(self, pipeline: &mut Pipeline, storage: &storage::ReadPlugin) { + pub fn spawn_stages(self, pipeline: &mut Pipeline, state: storage::ReadPlugin) { 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, storage, &mut cs_channel) - .expect("chainsync known-points should be defined"); + let cs_channel = transport.muxer.use_channel(5); pipeline.register_stage( "n2c", gasket::runtime::spawn_stage( - self::chainsync::Worker::new(cs_channel, 0, known_points, self.output), + self::chainsync::Worker::new( + cs_channel, + 0, + self.chain, + self.intersect, + state, + self.output, + ), gasket::runtime::Policy::default(), ), ); diff --git a/src/sources/n2n/blockfetch.rs b/src/sources/n2n/blockfetch.rs index 048450b9..04d0afc4 100644 --- a/src/sources/n2n/blockfetch.rs +++ b/src/sources/n2n/blockfetch.rs @@ -10,7 +10,7 @@ use crate::model::{ChainSyncCommand, ChainSyncCommandEx}; use crate::sources::utils; struct Observer<'a> { - output: &'a mut FanoutPort, + output: &'a mut OutputPort, } impl<'a> blockfetch::Observer for Observer<'a> { @@ -23,17 +23,17 @@ impl<'a> blockfetch::Observer for Observer<'a> { } pub type InputPort = gasket::messaging::InputPort; -pub type FanoutPort = gasket::messaging::FanoutPort; +pub type OutputPort = gasket::messaging::OutputPort; pub struct Worker { channel: Channel, block_count: gasket::metrics::Counter, input: InputPort, - output: FanoutPort, + output: OutputPort, } impl Worker { - pub fn new(channel: Channel, input: InputPort, output: FanoutPort) -> Self { + pub fn new(channel: Channel, input: InputPort, output: OutputPort) -> Self { Self { channel, input, diff --git a/src/sources/n2n/chainsync.rs b/src/sources/n2n/chainsync.rs index b258dee5..2ca21edb 100644 --- a/src/sources/n2n/chainsync.rs +++ b/src/sources/n2n/chainsync.rs @@ -10,7 +10,7 @@ use gasket::{ metrics::{Counter, Gauge}, }; -use crate::model::ChainSyncCommand; +use crate::{crosscut, model::ChainSyncCommand, sources::utils, storage}; use super::messages; @@ -101,8 +101,10 @@ type Runner = miniprotocols::Runner>; pub struct Worker { channel: Channel, pub min_depth: usize, - pub known_points: Option>, + chain: crosscut::ChainWellKnownInfo, + intersect: crosscut::IntersectConfig, //finalize_config: Option, + state: storage::ReadPlugin, runner: Cell>, output: OutputPort, block_count: gasket::metrics::Counter, @@ -113,13 +115,17 @@ impl Worker { pub fn new( channel: Channel, min_depth: usize, - known_points: Option>, + chain: crosscut::ChainWellKnownInfo, + intersect: crosscut::IntersectConfig, + state: storage::ReadPlugin, output: OutputPort, ) -> Self { Self { channel, min_depth, - known_points, + chain, + intersect, + state, output, runner: Cell::new(None), block_count: Default::default(), @@ -137,8 +143,18 @@ impl gasket::runtime::Worker for Worker { } fn bootstrap(&mut self) -> Result<(), gasket::error::Error> { + self.state.bootstrap().or_work_err()?; + + let known_points = utils::define_known_points( + &self.chain, + &self.intersect, + &mut self.state, + &mut self.channel, + ) + .or_work_err()?; + let mut runner = Runner::new(chainsync::Consumer::initial( - self.known_points.clone(), + known_points, ChainObserver::new( self.min_depth as usize, self.block_count.clone(), diff --git a/src/sources/n2n/mod.rs b/src/sources/n2n/mod.rs index 1cc38476..7c4e6555 100644 --- a/src/sources/n2n/mod.rs +++ b/src/sources/n2n/mod.rs @@ -7,7 +7,7 @@ use std::time::Duration; use gasket::{ error::AsWorkError, - messaging::{FanoutPort, InputPort, OutputPort}, + messaging::{InputPort, OutputPort}, retries, }; pub use messages::*; @@ -23,8 +23,6 @@ use crate::{ use self::transport::Transport; -use super::utils; - #[derive(Deserialize)] pub struct Config { pub address: String, @@ -49,7 +47,7 @@ pub struct Bootstrapper { config: Config, intersect: crosscut::IntersectConfig, chain: crosscut::ChainWellKnownInfo, - output: FanoutPort, + output: OutputPort, } impl Bootstrapper { @@ -66,22 +64,18 @@ impl Bootstrapper { ) } - pub fn borrow_output_port(&mut self) -> &'_ mut FanoutPort { + pub fn borrow_output_port(&mut self) -> &'_ mut OutputPort { &mut self.output } - pub fn spawn(self, pipeline: &mut Pipeline, storage: &storage::ReadPlugin) { + pub fn spawn_stages(self, pipeline: &mut Pipeline, state: storage::ReadPlugin) { let mut transport = self .bootstrap_transport() .expect("transport should be connected after several retries"); - let mut cs_channel = transport.muxer.use_channel(2); + let cs_channel = transport.muxer.use_channel(2); let bf_channel = transport.muxer.use_channel(3); - let known_points = - utils::define_known_points(&self.chain, &self.intersect, storage, &mut cs_channel) - .expect("chainsync known-points should be defined"); - let mut headers_out = OutputPort::::default(); let mut headers_in = InputPort::::default(); gasket::messaging::connect_ports(&mut headers_out, &mut headers_in, 10); @@ -89,7 +83,14 @@ impl Bootstrapper { pipeline.register_stage( "n2n-headers", gasket::runtime::spawn_stage( - self::chainsync::Worker::new(cs_channel, 0, known_points, headers_out), + self::chainsync::Worker::new( + cs_channel, + 0, + self.chain, + self.intersect, + state, + headers_out, + ), gasket::runtime::Policy::default(), ), ); diff --git a/src/sources/utils.rs b/src/sources/utils.rs index b646f319..09408e1c 100644 --- a/src/sources/utils.rs +++ b/src/sources/utils.rs @@ -58,7 +58,7 @@ pub fn find_end_of_chain( pub fn define_known_points( chain: &crosscut::ChainWellKnownInfo, intersect: &crosscut::IntersectConfig, - storage: &storage::ReadPlugin, + storage: &mut storage::ReadPlugin, channel: &mut Channel, ) -> Result>, crate::Error> { // if we have a cursor available, it should override any other configuration diff --git a/src/storage/mod.rs b/src/storage/mod.rs index b8c5569c..1eba9713 100644 --- a/src/storage/mod.rs +++ b/src/storage/mod.rs @@ -52,6 +52,12 @@ pub enum ReadPlugin { } impl ReadPlugin { + pub fn bootstrap(&mut self) -> Result<(), crate::Error> { + match self { + ReadPlugin::Redis(x) => x.bootstrap(), + } + } + pub fn read_state( &mut self, query: model::StateQuery, @@ -61,7 +67,7 @@ impl ReadPlugin { } } - pub fn read_cursor(&self) -> Result { + pub fn read_cursor(&mut self) -> Result { match self { ReadPlugin::Redis(x) => x.read_cursor(), } diff --git a/src/storage/redis.rs b/src/storage/redis.rs index 94053d70..635b4b3e 100644 --- a/src/storage/redis.rs +++ b/src/storage/redis.rs @@ -5,10 +5,13 @@ use gasket::{ runtime::{spawn_stage, WorkOutcome}, }; -use redis::Commands; +use redis::{Commands, Connection}; use serde::Deserialize; -use crate::{bootstrap, crosscut, model}; +use crate::{ + bootstrap, crosscut, + model::{self, StateData}, +}; type FunnelPort = gasket::messaging::FunnelPort; @@ -43,6 +46,7 @@ impl Bootstrapper { pub fn build_read_plugin(&self) -> ReadPlugin { ReadPlugin { config: self.config.clone(), + connection: None, } } @@ -103,6 +107,13 @@ impl gasket::runtime::Worker for Worker { .zadd(key, value, timestamp) .or_work_err()?; } + model::CRDTCommand::AnyWriteWins(key, value) => { + self.connection + .as_mut() + .unwrap() + .set(key, value) + .or_work_err()?; + } model::CRDTCommand::PNCounter(key, value) => { self.connection .as_mut() @@ -142,28 +153,58 @@ impl gasket::runtime::Worker for Worker { pub struct ReadPlugin { config: Config, + connection: Option, } impl ReadPlugin { - fn redis_connect(&self) -> Result { - let client = redis::Client::open(self.config.connection_params.clone())?; - client.get_connection() + pub fn bootstrap(&mut self) -> Result<(), crate::Error> { + let connection = redis::Client::open(self.config.connection_params.clone()) + .and_then(|c| c.get_connection()) + .map_err(crate::Error::storage)?; + + self.connection = Some(connection); + + Ok(()) } pub fn read_state( &mut self, query: model::StateQuery, ) -> Result { - match query { - model::StateQuery::LatestKeyValue(_) => todo!(), - model::StateQuery::SetMembers(_) => todo!(), - } - } + let value = match query { + model::StateQuery::KeyValue(key) => self + .connection + .as_mut() + .unwrap() + .get(key) + .map(|v| StateData::KeyValue(v)) + .map_err(crate::Error::storage)?, + model::StateQuery::LatestKeyValue(key) => self + .connection + .as_mut() + .unwrap() + .zrevrange(key, 0, 1) + .map(|v| StateData::KeyValue(v)) + .map_err(crate::Error::storage)?, + model::StateQuery::SetMembers(key) => self + .connection + .as_mut() + .unwrap() + .get(key) + .map(|v| StateData::SetMembers(v)) + .map_err(crate::Error::storage)?, + }; - pub fn read_cursor(&self) -> Result { - let mut connection = self.redis_connect().map_err(crate::Error::storage)?; + Ok(value.into()) + } - let raw: Option = connection.get("_cursor").map_err(crate::Error::storage)?; + pub fn read_cursor(&mut self) -> Result { + let raw: Option = self + .connection + .as_mut() + .unwrap() + .get("_cursor") + .map_err(crate::Error::storage)?; let point = match raw { Some(x) => Some(crosscut::PointArg::from_str(&x)?),