refactor: Move fallible tasks inside workers

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)

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),
 }

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 {

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")]

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))?;
 

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(())

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))?;
         }
 

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))?;
         }
 

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))?;
 

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);
         }
 

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<model::ChainSyncCommandEx> {
+    pub fn borrow_output_port(&mut self) -> &'_ mut OutputPort<model::ChainSyncCommandEx> {
         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),
         }
     }
 }

src/sources/n2c/chainsync.rs

@@ -11,26 +11,23 @@ use gasket::{
 };
 
 use crate::{
+    crosscut,
     model::{ChainSyncCommandEx, MultiEraBlock},
     sources::utils,
+    storage,
 };
 
 struct ChainObserver {
     min_depth: usize,
-    output: gasket::messaging::FanoutPort<ChainSyncCommandEx>,
+    output: OutputPort,
     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 {
+    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<chainsync::BlockContent> for ChainObserver {
     }
 }
 
-type OutputPort = gasket::messaging::FanoutPort<ChainSyncCommandEx>;
+type OutputPort = gasket::messaging::OutputPort<ChainSyncCommandEx>;
 type Runner = miniprotocols::Runner<chainsync::BlockConsumer<ChainObserver>>;
 
 pub struct Worker {
     channel: Channel,
     pub min_depth: usize,
-    pub known_points: Option<Vec<Point>>,
+    chain: crosscut::ChainWellKnownInfo,
+    intersect: crosscut::IntersectConfig,
+    state: storage::ReadPlugin,
+    output: OutputPort,
     //finalize_config: Option<FinalizeConfig>,
-    runner: Runner,
+    runner: Option<Runner>,
     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<Vec<Point>>,
+        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!(

src/sources/n2c/mod.rs

@@ -3,16 +3,14 @@ mod transport;
 
 use std::time::Duration;
 
-use gasket::{error::AsWorkError, messaging::FanoutPort, retries};
+use gasket::{error::AsWorkError, messaging::OutputPort, retries};
 
 use serde::Deserialize;
 
 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<ChainSyncCommandEx>,
+    output: OutputPort<ChainSyncCommandEx>,
 }
 
 impl Bootstrapper {
@@ -54,25 +52,28 @@ impl Bootstrapper {
         )
     }
 
-    pub fn borrow_output_port(&mut self) -> &'_ mut FanoutPort<ChainSyncCommandEx> {
+    pub fn borrow_output_port(&mut self) -> &'_ mut OutputPort<ChainSyncCommandEx> {
         &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(),
             ),
         );

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<ChainSyncCommand>;
-pub type FanoutPort = gasket::messaging::FanoutPort<ChainSyncCommandEx>;
+pub type OutputPort = gasket::messaging::OutputPort<ChainSyncCommandEx>;
 
 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,