attempt to re-create websocket creation on failure

Cargo.lock

@@ -6694,6 +6694,7 @@ dependencies = [
  "tendermint",
  "tendermint-rpc",
  "thiserror",
+ "time",
  "tokio",
  "tokio-stream",
  "tokio-util",

common/nyxd-scraper/Cargo.toml

@@ -21,6 +21,7 @@ sqlx = { workspace = true, features = ["runtime-tokio-rustls", "sqlite", "macros
 tendermint.workspace = true
 tendermint-rpc = { workspace = true, features = ["websocket-client", "http-client"] }
 thiserror.workspace = true
+time = { workspace = true }
 tokio = { workspace = true, features = ["full"] }
 tokio-stream = "0.1.14"
 tokio-util = { version = "0.7.10", features = ["rt"]}

common/nyxd-scraper/src/error.rs

@@ -84,7 +84,7 @@ pub enum ScraperError {
     EmptyBlockData { query: String },
 
     #[error("reached maximum number of allowed errors for subscription events")]
-    MaximumSubscriptionFailures,
+    MaximumWebSocketFailures,
 
     #[error("failed to begin storage tx: {source}")]
     StorageTxBeginFailure {

common/nyxd-scraper/src/scraper/mod.rs

@@ -6,11 +6,10 @@ use crate::block_requester::BlockRequester;
 use crate::error::ScraperError;
 use crate::modules::{BlockModule, MsgModule, TxModule};
 use crate::rpc_client::RpcClient;
-use crate::scraper::subscriber::{run_websocket_driver, ChainSubscriber};
+use crate::scraper::subscriber::ChainSubscriber;
 use crate::storage::ScraperStorage;
 use std::path::PathBuf;
 use std::sync::Arc;
-use tendermint_rpc::WebSocketClientDriver;
 use tokio::sync::mpsc::{channel, unbounded_channel};
 use tokio::sync::Notify;
 use tokio_util::sync::CancellationToken;
@@ -67,20 +66,15 @@ impl NyxdScraperBuilder {
         block_processor.set_tx_modules(self.tx_modules);
         block_processor.set_msg_modules(self.msg_modules);
 
-        let mut chain_subscriber = ChainSubscriber::new(
+        let chain_subscriber = ChainSubscriber::new(
             &scraper.config.websocket_url,
             scraper.cancel_token.clone(),
+            scraper.task_tracker.clone(),
             processing_tx,
         )
         .await?;
-        let ws_driver = chain_subscriber.ws_driver();
 
-        scraper.start_tasks(
-            block_requester,
-            block_processor,
-            chain_subscriber,
-            ws_driver,
-        );
+        scraper.start_tasks(block_requester, block_processor, chain_subscriber);
 
         Ok(scraper)
     }
@@ -141,7 +135,6 @@ impl NyxdScraper {
         mut block_requester: BlockRequester,
         mut block_processor: BlockProcessor,
         mut chain_subscriber: ChainSubscriber,
-        ws_driver: WebSocketClientDriver,
     ) {
         self.task_tracker
             .spawn(async move { block_requester.run().await });
@@ -149,8 +142,7 @@ impl NyxdScraper {
             .spawn(async move { block_processor.run().await });
         self.task_tracker
             .spawn(async move { chain_subscriber.run().await });
-        self.task_tracker
-            .spawn(run_websocket_driver(ws_driver, self.cancel_token.clone()));
+
         self.task_tracker.close();
     }
 
@@ -176,21 +168,16 @@ impl NyxdScraper {
             rpc_client,
         )
         .await?;
-        let mut chain_subscriber = ChainSubscriber::new(
+        let chain_subscriber = ChainSubscriber::new(
             &self.config.websocket_url,
             self.cancel_token.clone(),
+            self.task_tracker.clone(),
             processing_tx,
         )
         .await?;
-        let ws_driver = chain_subscriber.ws_driver();
 
         // spawn them
-        self.start_tasks(
-            block_requester,
-            block_processor,
-            chain_subscriber,
-            ws_driver,
-        );
+        self.start_tasks(block_requester, block_processor, chain_subscriber);
 
         Ok(())
     }

common/nyxd-scraper/src/scraper/subscriber.rs

@@ -6,18 +6,25 @@ use crate::error::ScraperError;
 use tendermint_rpc::event::Event;
 use tendermint_rpc::query::EventType;
 use tendermint_rpc::{SubscriptionClient, WebSocketClient, WebSocketClientDriver};
+use time::{Duration, OffsetDateTime};
 use tokio::sync::mpsc::UnboundedSender;
 use tokio_stream::StreamExt;
 use tokio_util::sync::CancellationToken;
+use tokio_util::task::TaskTracker;
 use tracing::{error, info, warn};
 use url::Url;
 
 const MAX_FAILURES: usize = 10;
+const MAX_RECONNECTION_ATTEMPTS: usize = 8;
+const SOCKET_FAILURE_RESET: Duration = Duration::hours(2);
 
 pub struct ChainSubscriber {
     cancel: CancellationToken,
+    task_tracker: TaskTracker,
+
     block_sender: UnboundedSender<BlockToProcess>,
 
+    websocket_endpoint: Url,
     websocket_client: WebSocketClient,
     websocket_driver: Option<WebSocketClientDriver>,
 }
@@ -26,6 +33,7 @@ impl ChainSubscriber {
     pub async fn new(
         websocket_endpoint: &Url,
         cancel: CancellationToken,
+        task_tracker: TaskTracker,
         block_sender: UnboundedSender<BlockToProcess>,
     ) -> Result<Self, ScraperError> {
         // sure, we could have just used websocket client entirely, but let's keep the logic for
@@ -39,7 +47,9 @@ impl ChainSubscriber {
 
         Ok(ChainSubscriber {
             cancel,
+            task_tracker,
             block_sender,
+            websocket_endpoint: websocket_endpoint.clone(),
             websocket_client: client,
             websocket_driver: Some(driver),
         })
@@ -53,8 +63,48 @@ impl ChainSubscriber {
         Ok(())
     }
 
-    pub(crate) async fn run(&mut self) -> Result<(), ScraperError> {
-        let _drop_guard = self.cancel.clone().drop_guard();
+    async fn remake_connection(&mut self) -> Result<(), ScraperError> {
+        info!(
+            "attempting to reestablish connection to {}",
+            self.websocket_endpoint
+        );
+
+        let (client, driver) = WebSocketClient::new(self.websocket_endpoint.as_str())
+            .await
+            .map_err(|source| ScraperError::WebSocketConnectionFailure {
+                url: self.websocket_endpoint.to_string(),
+                source,
+            })?;
+        self.websocket_client = client;
+        self.websocket_driver = Some(driver);
+
+        info!(
+            "managed to reestablish the websocket connection to {}",
+            self.websocket_endpoint
+        );
+        Ok(())
+    }
+
+    /// Returns whether the method exited due to the cancellation
+    async fn run_chain_subscription(&mut self) -> Result<bool, ScraperError> {
+        let Some(ws_driver) = self.websocket_driver.take() else {
+            error!("the websocket driver hasn't been created - we probably failed to establish the connection");
+            return Ok(false);
+        };
+
+        let driver_cancel = CancellationToken::new();
+        let _driver_guard = driver_cancel.clone().drop_guard();
+
+        // spawn the websocket driver task
+        let driver_handle = {
+            self.task_tracker.reopen();
+            let handle = self
+                .task_tracker
+                .spawn(run_websocket_driver(ws_driver, driver_cancel));
+            self.task_tracker.close();
+            handle
+        };
+        tokio::pin!(driver_handle);
 
         info!("creating chain subscription");
         let mut subs = self
@@ -70,12 +120,17 @@ impl ChainSubscriber {
             tokio::select! {
                 _ = self.cancel.cancelled() => {
                     info!("received cancellation token");
-                    break
+                    // note: `_driver_guard` will get dropped here thus causing cancellation of the driver task
+                    return Ok(true)
+                }
+                _ = &mut driver_handle => {
+                    error!("our websocket driver has finished execution");
+                    return Ok(self.cancel.is_cancelled())
                 }
                 maybe_event = subs.next() => {
                     let Some(maybe_event) = maybe_event else {
                         warn!("stopped receiving new events");
-                        break;
+                        return Ok(false)
                     };
                     match maybe_event {
                         Ok(event) => {
@@ -92,38 +147,95 @@ impl ChainSubscriber {
                         }
                     }
                     if failures >= MAX_FAILURES {
-                        // note: the drop_guard will get dropped and thus cause a shutdown
-                        return Err(ScraperError::MaximumSubscriptionFailures);
+                        return Ok(false)
                     }
                 }
             }
         }
-        Ok(())
     }
 
-    pub(crate) fn ws_driver(&mut self) -> WebSocketClientDriver {
-        #[allow(clippy::expect_used)]
-        self.websocket_driver
-            .take()
-            .expect("websocket driver has already been started!")
+    async fn websocket_backoff(&mut self, failure_count: usize) -> bool {
+        const MINIMUM_WAIT_MS: u64 = 10_000;
+        const INCREMENTAL_WAIT_MS: u64 = 30_000;
+
+        let backoff_duration_ms = MINIMUM_WAIT_MS + INCREMENTAL_WAIT_MS * failure_count as u64;
+        info!("going to wait {backoff_duration_ms} ms before re-attempting the reconnection");
+
+        tokio::select! {
+            _ = self.cancel.cancelled() => {
+                info!("received cancellation token");
+                true
+            }
+            _ = tokio::time::sleep(std::time::Duration::from_millis(backoff_duration_ms)) => false,
+        }
+    }
+
+    pub(crate) async fn run(&mut self) -> Result<(), ScraperError> {
+        let _drop_guard = self.cancel.clone().drop_guard();
+        let mut socket_failures = 0;
+        let mut last_failure = OffsetDateTime::now_utc();
+
+        loop {
+            if self.cancel.is_cancelled() {
+                return Ok(());
+            }
+
+            match self.run_chain_subscription().await {
+                Ok(cancelled) => {
+                    if cancelled {
+                        // we're in the middle of a shutdown
+                        return Ok(());
+                    }
+                    socket_failures += 1;
+                }
+                Err(err) => {
+                    error!("failed to create chain subscription: {err}");
+                    socket_failures += 1;
+                }
+            }
+
+            warn!("current socket failure count: {socket_failures}. the last failure was at {last_failure}");
+
+            let now = OffsetDateTime::now_utc();
+
+            // if it's been a while since the last failure, reset the count
+            if now - last_failure > SOCKET_FAILURE_RESET {
+                warn!("resetting the failure count to 1");
+                socket_failures = 1;
+            }
+            last_failure = now;
+
+            if socket_failures >= MAX_RECONNECTION_ATTEMPTS {
+                error!("reached the maximum allowed failure count");
+                return Err(ScraperError::MaximumWebSocketFailures);
+            }
+
+            // BACKOFF
+            let cancelled = self.websocket_backoff(socket_failures).await;
+            if cancelled {
+                return Ok(());
+            }
+
+            if let Err(err) = self.remake_connection().await {
+                error!("failed to re-establish the websocket connection: {err}");
+            }
+        }
     }
 }
 
-pub async fn run_websocket_driver(driver: WebSocketClientDriver, cancel: CancellationToken) {
+pub async fn run_websocket_driver(driver: WebSocketClientDriver, driver_cancel: CancellationToken) {
     info!("starting websocket driver");
     tokio::select! {
-        _ = cancel.cancelled() => {
+        _ = driver_cancel.cancelled() => {
             info!("received cancellation token")
         }
         res = driver.run() => {
             match res {
                 Ok(_) => info!("our websocket driver has finished execution"),
                 Err(err) => {
-                    // TODO: in the future just attempt to reconnect
                     error!("our websocket driver has errored out: {err}");
                 }
             }
-            cancel.cancel()
         }
     }
 }

nym-validator-rewarder/src/rewarder/mod.rs

@@ -95,6 +95,10 @@ impl Rewarder {
                 return Err(NymRewarderError::EmptyBlockSigningWhitelist);
             }
 
+            if config.block_signing.monitor_only {
+                info!("the block signing rewarding is running in monitor only mode");
+            }
+
             let nyxd_scraper = NyxdScraper::new(config.scraper_config()).await?;
 
             Some(EpochSigning {