feat: Implement automatic source reconnect

Cargo.toml

@@ -27,7 +27,7 @@ serde = { version = "1.0.136", features = ["derive"] }
 serde_json = "1.0.79"
 minicbor = "0.14.1"
 prometheus_exporter = { version = "0.8.4", default-features = false }
-# gasket = { path = "../../gasketlibs/gasket-rs" }
+# gasket = { path = "../../construkts/gasket-rs" }
 gasket = { git = "https://github.com/construkts/gasket-rs.git" }
 thiserror = "1.0.30"
 redis = "0.21.5"

src/bin/scrolls/console.rs

@@ -74,7 +74,7 @@ impl TuiConsole {
     }
 
     fn refresh(&self, pipeline: &Pipeline) {
-        for (stage, tether) in pipeline.tethers.iter() {
+        for tether in pipeline.tethers.iter() {
             let state = match tether.check_state() {
                 gasket::runtime::TetherState::Dropped => "dropped!",
                 gasket::runtime::TetherState::Blocked(_) => "blocked!",
@@ -90,7 +90,7 @@ impl TuiConsole {
             match tether.read_metrics() {
                 Ok(readings) => {
                     for (key, value) in readings {
-                        match (*stage, key, value) {
+                        match (tether.name(), key, value) {
                             (_, "chain_tip", Reading::Gauge(x)) => {
                                 self.chainsync_progress.set_length(x as u64);
                             }
@@ -173,24 +173,24 @@ impl PlainConsole {
             return;
         }
 
-        for (stage, tether) in pipeline.tethers.iter() {
+        for tether in pipeline.tethers.iter() {
             match tether.check_state() {
                 gasket::runtime::TetherState::Dropped => {
-                    log::error!("[{}] stage tether has been dropped", stage);
+                    log::error!("[{}] stage tether has been dropped", tether.name());
                 }
                 gasket::runtime::TetherState::Blocked(_) => {
-                    log::warn!("[{}] stage tehter is blocked or not reporting state", stage);
+                    log::warn!("[{}] stage tehter is blocked or not reporting state", tether.name());
                 }
                 gasket::runtime::TetherState::Alive(state) => {
-                    log::debug!("[{}] stage is alive with state: {:?}", stage, state);
+                    log::debug!("[{}] stage is alive with state: {:?}", tether.name(), state);
                     match tether.read_metrics() {
                         Ok(readings) => {
                             for (key, value) in readings {
-                                log::debug!("[{}] metric `{}` = {:?}", stage, key, value);
+                                log::debug!("[{}] metric `{}` = {:?}", tether.name(), key, value);
                             }
                         }
                         Err(err) => {
-                            log::error!("[{}] error reading metrics: {}", stage, err)
+                            log::error!("[{}] error reading metrics: {}", tether.name(), err)
                         }
                     }
                 }

src/bin/scrolls/daemon.rs

@@ -66,7 +66,7 @@ fn should_stop(pipeline: &bootstrap::Pipeline) -> bool {
     pipeline
         .tethers
         .iter()
-        .any(|(_, tether)| match tether.check_state() {
+        .any(|tether| match tether.check_state() {
             gasket::runtime::TetherState::Alive(x) => match x {
                 gasket::runtime::StageState::StandBy => true,
                 _ => false,
@@ -76,9 +76,9 @@ fn should_stop(pipeline: &bootstrap::Pipeline) -> bool {
 }
 
 fn shutdown(pipeline: bootstrap::Pipeline) {
-    for (name, tether) in pipeline.tethers {
+    for tether in pipeline.tethers {
         let state = tether.check_state();
-        log::warn!("dismissing stage: {} with state {:?}", name, state);
+        log::warn!("dismissing stage: {} with state {:?}", tether.name(), state);
         tether.dismiss_stage().expect("stage stops");
 
         // Can't join the stage because there's a risk of deadlock, usually

src/bootstrap.rs

@@ -2,10 +2,8 @@ use crate::{enrich, reducers, sources, storage};
 
 use gasket::{messaging::connect_ports, runtime::Tether};
 
-type NamedTether = (&'static str, Tether);
-
 pub struct Pipeline {
-    pub tethers: Vec<NamedTether>,
+    pub tethers: Vec<Tether>,
 }
 
 impl Pipeline {
@@ -15,8 +13,8 @@ impl Pipeline {
         }
     }
 
-    pub fn register_stage(&mut self, name: &'static str, tether: Tether) {
-        self.tethers.push((name, tether));
+    pub fn register_stage(&mut self, tether: Tether) {
+        self.tethers.push(tether);
     }
 }
 

src/enrich/skip.rs

@@ -40,13 +40,13 @@ impl Bootstrapper {
         };
 
         pipeline.register_stage(
-            "enrich-skip",
             spawn_stage(
                 worker,
                 gasket::runtime::Policy {
                     tick_timeout: Some(Duration::from_secs(600)),
                     ..Default::default()
                 },
+                Some("enrich-skip"),
             ),
         );
     }

src/enrich/sled.rs

@@ -69,13 +69,13 @@ impl Bootstrapper {
         };
 
         pipeline.register_stage(
-            "enrich-sled",
             spawn_stage(
                 worker,
                 gasket::runtime::Policy {
                     tick_timeout: Some(Duration::from_secs(600)),
                     ..Default::default()
                 },
+                Some("enrich-sled"),
             ),
         );
     }

src/reducers/mod.rs

@@ -129,13 +129,13 @@ impl Bootstrapper {
     pub fn spawn_stages(self, pipeline: &mut bootstrap::Pipeline) {
         let worker = worker::Worker::new(self.reducers, self.input, self.output, self.policy);
         pipeline.register_stage(
-            "reducers",
             spawn_stage(
                 worker,
                 gasket::runtime::Policy {
                     tick_timeout: Some(Duration::from_secs(600)),
                     ..Default::default()
                 },
+                Some("reducers"),
             ),
         );
     }

src/sources/n2c/chainsync.rs

@@ -2,7 +2,6 @@ use std::{collections::HashMap, ops::Deref};
 
 use pallas::ledger::traverse::MultiEraBlock;
 use pallas::network::miniprotocols::{self, chainsync, Agent, Point};
-use pallas::network::multiplexer;
 
 use gasket::{
     error::AsWorkError,
@@ -11,6 +10,8 @@ use gasket::{
 
 use crate::{crosscut, model::RawBlockPayload, sources::utils};
 
+use super::transport::Transport;
+
 struct ChainObserver {
     min_depth: usize,
     output: OutputPort,
@@ -97,35 +98,37 @@ type OutputPort = gasket::messaging::OutputPort<RawBlockPayload>;
 type MyAgent = chainsync::BlockConsumer<ChainObserver>;
 
 pub struct Worker {
-    channel: multiplexer::StdChannelBuffer,
+    socket: String,
     min_depth: usize,
     chain: crosscut::ChainWellKnownInfo,
     intersect: crosscut::IntersectConfig,
     cursor: crosscut::Cursor,
     //finalize_config: Option<FinalizeConfig>,
     agent: Option<MyAgent>,
+    transport: Option<Transport>,
     output: OutputPort,
     block_count: gasket::metrics::Counter,
     chain_tip: gasket::metrics::Gauge,
 }
 
 impl Worker {
     pub fn new(
-        channel: multiplexer::StdChannelBuffer,
+        socket: String,
         min_depth: usize,
         chain: crosscut::ChainWellKnownInfo,
         intersect: crosscut::IntersectConfig,
         cursor: crosscut::Cursor,
         output: OutputPort,
     ) -> Self {
         Self {
-            channel,
+            socket,
             min_depth,
             chain,
             intersect,
             cursor,
             output,
             agent: None,
+            transport: None,
             block_count: Default::default(),
             chain_tip: Default::default(),
         }
@@ -141,11 +144,13 @@ impl gasket::runtime::Worker for Worker {
     }
 
     fn bootstrap(&mut self) -> Result<(), gasket::error::Error> {
+        let mut transport = Transport::setup(&self.socket, self.chain.magic).or_retry()?;
+
         let known_points = utils::define_known_points(
             &self.chain,
             &self.intersect,
             &self.cursor,
-            &mut self.channel,
+            &mut transport.channel5,
         )
         .or_retry()?;
 
@@ -162,18 +167,21 @@ impl gasket::runtime::Worker for Worker {
         .or_retry()?;
 
         self.agent = Some(agent);
+        self.transport = Some(transport);
 
         Ok(())
     }
 
     fn work(&mut self) -> gasket::runtime::WorkResult {
         let agent = self.agent.take().unwrap();
+        let mut transport = self.transport.take().unwrap();
 
-        let agent = miniprotocols::run_agent_step(agent, &mut self.channel).or_restart()?;
+        let agent = miniprotocols::run_agent_step(agent, &mut transport.channel5).or_restart()?;
 
         let is_done = agent.is_done();
 
         self.agent = Some(agent);
+        self.transport = Some(transport);
 
         match is_done {
             true => Ok(gasket::runtime::WorkOutcome::Done),

src/sources/n2c/mod.rs

@@ -4,9 +4,8 @@ mod transport;
 use serde::Deserialize;
 use std::time::Duration;
 
-use self::transport::Transport;
 use crate::{bootstrap::Pipeline, crosscut, model::RawBlockPayload};
-use gasket::{error::AsWorkError, messaging::OutputPort, retries};
+use gasket::messaging::OutputPort;
 
 #[derive(Deserialize)]
 pub struct Config {
@@ -36,45 +35,31 @@ pub struct Bootstrapper {
 }
 
 impl Bootstrapper {
-    fn bootstrap_transport(&self) -> Result<Transport, crate::Error> {
-        gasket::retries::retry_operation(
-            || Transport::setup(&self.config.path, self.chain.magic).or_retry(),
-            &retries::Policy {
-                max_retries: 5,
-                backoff_factor: 2,
-                backoff_unit: Duration::from_secs(1),
-                max_backoff: Duration::from_secs(60),
-            },
-            None,
-        )
-        .map_err(crate::Error::source)
-    }
-
     pub fn borrow_output_port(&mut self) -> &'_ mut OutputPort<RawBlockPayload> {
         &mut self.output
     }
 
     pub fn spawn_stages(self, pipeline: &mut Pipeline, cursor: &Option<crosscut::PointArg>) {
-        let transport = self
-            .bootstrap_transport()
-            .expect("transport should be connected after several retries");
-
-        pipeline.register_stage(
-            "n2c",
-            gasket::runtime::spawn_stage(
-                self::chainsync::Worker::new(
-                    transport.channel5,
-                    0,
-                    self.chain,
-                    self.intersect,
-                    cursor.clone(),
-                    self.output,
-                ),
-                gasket::runtime::Policy {
-                    tick_timeout: Some(Duration::from_secs(600)),
-                    ..Default::default()
-                },
+        pipeline.register_stage(gasket::runtime::spawn_stage(
+            self::chainsync::Worker::new(
+                self.config.path.clone(),
+                0,
+                self.chain,
+                self.intersect,
+                cursor.clone(),
+                self.output,
             ),
-        );
+            gasket::runtime::Policy {
+                tick_timeout: Some(Duration::from_secs(600)),
+                bootstrap_retry: gasket::retries::Policy {
+                    max_retries: 20,
+                    backoff_factor: 2,
+                    backoff_unit: Duration::from_secs(1),
+                    max_backoff: Duration::from_secs(60),
+                },
+                ..Default::default()
+            },
+            Some("n2c"),
+        ));
     }
 }