feat(supervisor/core): implemented task retrier

crates/supervisor/core/src/logindexer/indexer.rs

@@ -141,9 +141,9 @@ pub enum LogIndexerError {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::syncnode::{AuthenticationError, ClientError};
     use alloy_primitives::{Address, B256, Bytes};
     use async_trait::async_trait;
-    use jsonrpsee::core::ClientError;
     use kona_interop::{ExecutingMessageBuilder, InteropProvider, SuperchainBuilder};
     use kona_protocol::BlockInfo;
     use kona_supervisor_storage::StorageError;
@@ -271,7 +271,11 @@ mod tests {
 
         let mut mock_provider = MockBlockProvider::new();
         mock_provider.expect_fetch_receipts().withf(move |hash| *hash == block_hash).returning(
-            |_| Err(ManagedNodeError::Client(ClientError::Custom("forced error".to_string()))),
+            |_| {
+                Err(ManagedNodeError::Client(ClientError::Authentication(
+                    AuthenticationError::InvalidHeader,
+                )))
+            },
         );
 
         mock_provider.expect_block_by_number().returning(|_| Ok(BlockInfo::default())); // Not used

crates/supervisor/core/src/supervisor.rs

@@ -241,7 +241,10 @@
             let provider = RootProvider::<Ethereum>::new_http(url);
             let client = Arc::new(Client::new(config.clone()));
 
-            let chain_id = client.chain_id().await?;
+            let chain_id = client.chain_id().await.map_err(|err| {
+                error!(target: "supervisor_service", %err, "Failed to get chain ID from client");
+                SupervisorError::Initialise("failed to get chain id from client".to_string())
+            })?;
             let db = self.database_factory.get_db(chain_id)?;
 
             let managed_node = ManagedNode::<ChainDb, Client>::new(

crates/supervisor/core/src/syncnode/client.rs

@@ -1,4 +1,4 @@
-use super::{AuthenticationError, ManagedNodeError, metrics::Metrics};
+use super::{AuthenticationError, ClientError, metrics::Metrics};
 use alloy_primitives::{B256, ChainId};
 use alloy_rpc_types_engine::{Claims, JwtSecret};
 use alloy_rpc_types_eth::BlockNumHash;
@@ -21,31 +21,26 @@
 #[async_trait]
 pub trait ManagedNodeClient: Debug {
     /// Returns the [`ChainId`] of the managed node.
-    async fn chain_id(&self) -> Result<ChainId, ManagedNodeError>;
+    async fn chain_id(&self) -> Result<ChainId, ClientError>;
 
     /// Subscribes to [`SubscriptionEvent`] from the managed node.
-    async fn subscribe_events(&self) -> Result<Subscription<SubscriptionEvent>, ManagedNodeError>;
+    async fn subscribe_events(&self) -> Result<Subscription<SubscriptionEvent>, ClientError>;
 
     /// Fetches [`Receipts`] for a given block hash.
-    async fn fetch_receipts(&self, block_hash: B256) -> Result<Receipts, ManagedNodeError>;
+    async fn fetch_receipts(&self, block_hash: B256) -> Result<Receipts, ClientError>;
 
     /// Fetches the [`OutputV0`] at a specific timestamp.
-    async fn output_v0_at_timestamp(&self, timestamp: u64) -> Result<OutputV0, ManagedNodeError>;
+    async fn output_v0_at_timestamp(&self, timestamp: u64) -> Result<OutputV0, ClientError>;
 
     /// Fetches the pending [`OutputV0`] at a specific timestamp.
-    async fn pending_output_v0_at_timestamp(
-        &self,
-        timestamp: u64,
-    ) -> Result<OutputV0, ManagedNodeError>;
+    async fn pending_output_v0_at_timestamp(&self, timestamp: u64)
+    -> Result<OutputV0, ClientError>;
 
     /// Fetches the L2 [`BlockInfo`] by timestamp.
-    async fn l2_block_ref_by_timestamp(
-        &self,
-        timestamp: u64,
-    ) -> Result<BlockInfo, ManagedNodeError>;
+    async fn l2_block_ref_by_timestamp(&self, timestamp: u64) -> Result<BlockInfo, ClientError>;
 
     /// Fetches the [`BlockInfo`] by block number.
-    async fn block_ref_by_number(&self, block_number: u64) -> Result<BlockInfo, ManagedNodeError>;
+    async fn block_ref_by_number(&self, block_number: u64) -> Result<BlockInfo, ClientError>;
 
     /// Resets the node state with the provided block IDs.
     async fn reset(
@@ -55,29 +50,29 @@
         local_safe_id: BlockNumHash,
         cross_safe_id: BlockNumHash,
         finalised_id: BlockNumHash,
-    ) -> Result<(), ManagedNodeError>;
+    ) -> Result<(), ClientError>;
 
     /// Provides L1 [`BlockInfo`] to the managed node.
-    async fn provide_l1(&self, block_info: BlockInfo) -> Result<(), ManagedNodeError>;
+    async fn provide_l1(&self, block_info: BlockInfo) -> Result<(), ClientError>;
 
     /// Updates the finalized block ID in the managed node.
-    async fn update_finalized(
-        &self,
-        finalized_block_id: BlockNumHash,
-    ) -> Result<(), ManagedNodeError>;
+    async fn update_finalized(&self, finalized_block_id: BlockNumHash) -> Result<(), ClientError>;
 
     /// Updates the cross-unsafe block ID in the managed node.
     async fn update_cross_unsafe(
         &self,
         cross_unsafe_block_id: BlockNumHash,
-    ) -> Result<(), ManagedNodeError>;
+    ) -> Result<(), ClientError>;
 
     /// Updates the cross-safe block ID in the managed node.
     async fn update_cross_safe(
         &self,
         source_block_id: BlockNumHash,
         derived_block_id: BlockNumHash,
-    ) -> Result<(), ManagedNodeError>;
+    ) -> Result<(), ClientError>;
+
+    /// Resets the ws-client to None when server disconnects
+    async fn reset_ws_client(&self);
 }
 
 /// [`ClientConfig`] sets the configuration for the managed node client.
@@ -134,7 +129,7 @@
     }
 
     /// Creates authentication headers using JWT secret.
-    fn create_auth_headers(&self) -> Result<HeaderMap, ManagedNodeError> {
+    fn create_auth_headers(&self) -> Result<HeaderMap, ClientError> {
         let Some(jwt_secret) = self.config.jwt_secret() else {
             error!(target: "managed_node", "JWT secret not found or invalid");
             return Err(AuthenticationError::InvalidJwt.into())
@@ -163,7 +158,7 @@
 
     /// Returns a reference to the WebSocket client, creating it if it doesn't exist.
     // todo: support http client as well
-    pub async fn get_ws_client(&self) -> Result<Arc<WsClient>, ManagedNodeError> {
+    pub async fn get_ws_client(&self) -> Result<Arc<WsClient>, ClientError> {
         let mut ws_client_guard = self.ws_client.lock().await;
         if ws_client_guard.is_none() {
             let headers = self.create_auth_headers().inspect_err(|err| {
@@ -182,7 +177,13 @@
 
 #[async_trait]
 impl ManagedNodeClient for Client {
-    async fn chain_id(&self) -> Result<ChainId, ManagedNodeError> {
+    async fn reset_ws_client(&self) {
+        let mut ws_client_guard = self.ws_client.lock().await;
+        if ws_client_guard.is_some() {
+            *ws_client_guard = None;
+        };
+    }
+    async fn chain_id(&self) -> Result<ChainId, ClientError> {
         if let Some(chain_id) = self.chain_id.get() {
             return Ok(*chain_id);
         }
@@ -210,7 +211,7 @@
         Ok(chain_id)
     }
 
-    async fn subscribe_events(&self) -> Result<Subscription<SubscriptionEvent>, ManagedNodeError> {
+    async fn subscribe_events(&self) -> Result<Subscription<SubscriptionEvent>, ClientError> {
         let client = self.get_ws_client().await?; // This returns ManagedNodeError, handled by your function
         let subscription = observe_metrics_for_result_async!(
             Metrics::MANAGED_NODE_RPC_REQUESTS_SUCCESS_TOTAL,
@@ -226,7 +227,7 @@
         Ok(subscription)
     }
 
-    async fn fetch_receipts(&self, block_hash: B256) -> Result<Receipts, ManagedNodeError> {
+    async fn fetch_receipts(&self, block_hash: B256) -> Result<Receipts, ClientError> {
         let client = self.get_ws_client().await?; // This returns ManagedNodeError, handled by your function
         let receipts = observe_metrics_for_result_async!(
             Metrics::MANAGED_NODE_RPC_REQUESTS_SUCCESS_TOTAL,
@@ -242,7 +243,7 @@
         Ok(receipts)
     }
 
-    async fn output_v0_at_timestamp(&self, timestamp: u64) -> Result<OutputV0, ManagedNodeError> {
+    async fn output_v0_at_timestamp(&self, timestamp: u64) -> Result<OutputV0, ClientError> {
         let client = self.get_ws_client().await?;
         let output_v0 = observe_metrics_for_result_async!(
             Metrics::MANAGED_NODE_RPC_REQUESTS_SUCCESS_TOTAL,
@@ -261,7 +262,7 @@
     async fn pending_output_v0_at_timestamp(
         &self,
         timestamp: u64,
-    ) -> Result<OutputV0, ManagedNodeError> {
+    ) -> Result<OutputV0, ClientError> {
         let client = self.get_ws_client().await?;
         let output_v0 = observe_metrics_for_result_async!(
             Metrics::MANAGED_NODE_RPC_REQUESTS_SUCCESS_TOTAL,
@@ -277,10 +278,7 @@
         Ok(output_v0)
     }
 
-    async fn l2_block_ref_by_timestamp(
-        &self,
-        timestamp: u64,
-    ) -> Result<BlockInfo, ManagedNodeError> {
+    async fn l2_block_ref_by_timestamp(&self, timestamp: u64) -> Result<BlockInfo, ClientError> {
         let client = self.get_ws_client().await?;
         let block_info = observe_metrics_for_result_async!(
             Metrics::MANAGED_NODE_RPC_REQUESTS_SUCCESS_TOTAL,
@@ -296,7 +294,7 @@
         Ok(block_info)
     }
 
-    async fn block_ref_by_number(&self, block_number: u64) -> Result<BlockInfo, ManagedNodeError> {
+    async fn block_ref_by_number(&self, block_number: u64) -> Result<BlockInfo, ClientError> {
         let client = self.get_ws_client().await?;
         let block_info = observe_metrics_for_result_async!(
             Metrics::MANAGED_NODE_RPC_REQUESTS_SUCCESS_TOTAL,
@@ -319,7 +317,7 @@
         local_safe_id: BlockNumHash,
         cross_safe_id: BlockNumHash,
         finalised_id: BlockNumHash,
-    ) -> Result<(), ManagedNodeError> {
+    ) -> Result<(), ClientError> {
         let client = self.get_ws_client().await?;
         observe_metrics_for_result_async!(
             Metrics::MANAGED_NODE_RPC_REQUESTS_SUCCESS_TOTAL,
@@ -334,7 +332,7 @@
         Ok(())
     }
 
-    async fn provide_l1(&self, block_info: BlockInfo) -> Result<(), ManagedNodeError> {
+    async fn provide_l1(&self, block_info: BlockInfo) -> Result<(), ClientError> {
         let client = self.get_ws_client().await?;
         observe_metrics_for_result_async!(
             Metrics::MANAGED_NODE_RPC_REQUESTS_SUCCESS_TOTAL,
@@ -349,10 +347,7 @@
         Ok(())
     }
 
-    async fn update_finalized(
-        &self,
-        finalized_block_id: BlockNumHash,
-    ) -> Result<(), ManagedNodeError> {
+    async fn update_finalized(&self, finalized_block_id: BlockNumHash) -> Result<(), ClientError> {
         let client = self.get_ws_client().await?;
         observe_metrics_for_result_async!(
             Metrics::MANAGED_NODE_RPC_REQUESTS_SUCCESS_TOTAL,
@@ -370,7 +365,7 @@
     async fn update_cross_unsafe(
         &self,
         cross_unsafe_block_id: BlockNumHash,
-    ) -> Result<(), ManagedNodeError> {
+    ) -> Result<(), ClientError> {
         let client = self.get_ws_client().await?;
         observe_metrics_for_result_async!(
             Metrics::MANAGED_NODE_RPC_REQUESTS_SUCCESS_TOTAL,
@@ -389,7 +384,7 @@
         &self,
         source_block_id: BlockNumHash,
         derived_block_id: BlockNumHash,
-    ) -> Result<(), ManagedNodeError> {
+    ) -> Result<(), ClientError> {
         let client = self.get_ws_client().await?;
         observe_metrics_for_result_async!(
             Metrics::MANAGED_NODE_RPC_REQUESTS_SUCCESS_TOTAL,

crates/supervisor/core/src/syncnode/error.rs

@@ -3,44 +3,21 @@
 use thiserror::Error;
 
 /// Represents various errors that can occur during node management,
-#[derive(Debug, Error)]
+#[derive(Debug, Error, PartialEq, Eq)]
 pub enum ManagedNodeError {
     /// Represents an error that occurred while starting the managed node.
     #[error(transparent)]
-    Client(#[from] jsonrpsee::core::ClientError),
-
-    /// Represents an error that occurred while parsing a chain ID from a string.
-    #[error(transparent)]
-    ChainIdParseError(#[from] std::num::ParseIntError),
+    Client(#[from] ClientError),
 
     /// Represents an error that occurred while subscribing to the managed node.
     #[error("subscription error: {0}")]
     Subscription(#[from] SubscriptionError),
 
-    /// Represents an error that occurred while authenticating to the managed node.
-    #[error("failed to authenticate: {0}")]
-    Authentication(#[from] AuthenticationError),
-
     /// Represents an error that occurred while fetching data from the storage.
     #[error(transparent)]
     StorageError(#[from] StorageError),
 }
 
-impl PartialEq for ManagedNodeError {
-    fn eq(&self, other: &Self) -> bool {
-        use ManagedNodeError::*;
-        match (self, other) {
-            (Client(a), Client(b)) => a.to_string() == b.to_string(),
-            (ChainIdParseError(a), ChainIdParseError(b)) => a == b,
-            (Subscription(a), Subscription(b)) => a == b,
-            (Authentication(a), Authentication(b)) => a == b,
-            _ => false,
-        }
-    }
-}
-
-impl Eq for ManagedNodeError {}
-
 /// Error establishing authenticated connection to managed node.
 #[derive(Debug, Error, PartialEq, Eq)]
 pub enum AuthenticationError {
@@ -61,8 +38,12 @@
 }
 
 /// Error handling managed event task.
-#[derive(Debug, Error, PartialEq)]
+#[derive(Debug, Error, PartialEq, Eq)]
 pub enum ManagedEventTaskError {
+    /// Represents an error that occurred while starting the managed node.
+    #[error(transparent)]
+    Client(#[from] ClientError),
+
     /// Unable to successfully fetch next L1 block.
     #[error("failed to get block by number, number: {0}")]
     GetBlockByNumberFailed(u64),
@@ -78,10 +59,6 @@
         parent: B256,
     },
 
-    /// This should never happen, new() always sets the rpc client when creating the task.
-    #[error("rpc client for managed node is not set")]
-    ManagedNodeClientMissing,
-
     /// Managed node api call failed.
     #[error("managed node api call failed")]
     ManagedNodeAPICallFailed,
@@ -90,3 +67,33 @@
     #[error(transparent)]
     StorageError(#[from] StorageError),
 }
+
+/// Represents errors that can occur while interacting with the managed node client.
+#[derive(Debug, Error)]
+pub enum ClientError {
+    /// Represents an error that occurred while starting the managed node.
+    #[error(transparent)]
+    Client(#[from] jsonrpsee::core::ClientError),
+
+    /// Represents an error that occurred while authenticating to the managed node.
+    #[error("failed to authenticate: {0}")]
+    Authentication(#[from] AuthenticationError),
+
+    /// Represents an error that occurred while parsing a chain ID from a string.
+    #[error(transparent)]
+    ChainIdParseError(#[from] std::num::ParseIntError),
+}
+
+impl PartialEq for ClientError {
+    fn eq(&self, other: &Self) -> bool {
+        use ClientError::*;
+        match (self, other) {
+            (Client(a), Client(b)) => a.to_string() == b.to_string(),
+            (Authentication(a), Authentication(b)) => a == b,
+            (ChainIdParseError(a), ChainIdParseError(b)) => a == b,
+            _ => false,
+        }
+    }
+}
+
+impl Eq for ClientError {}

crates/supervisor/core/src/syncnode/mod.rs

@@ -5,7 +5,9 @@ mod node;
 pub use node::ManagedNode;
 
 mod error;
-pub use error::{AuthenticationError, ManagedEventTaskError, ManagedNodeError, SubscriptionError};
+pub use error::{
+    AuthenticationError, ClientError, ManagedEventTaskError, ManagedNodeError, SubscriptionError,
+};
 
 mod traits;
 pub use traits::{
@@ -19,3 +21,4 @@ pub use client::{Client, ClientConfig, ManagedNodeClient};
 pub(super) mod metrics;
 pub(super) mod resetter;
 pub(super) mod task;
+pub(super) mod utils;