fix(supervisor/core): derivation reset

crates/supervisor/core/src/chain_processor/chain.rs

@@ -117,7 +117,10 @@
     use super::*;
     use crate::{
         event::ChainEvent,
-        syncnode::{ManagedNodeApiProvider, ManagedNodeError, NodeSubscriber, ReceiptProvider},
+        syncnode::{
+            ManagedNodeController, ManagedNodeDataProvider, ManagedNodeError, NodeSubscriber,
+            ReceiptProvider,
+        },
     };
     use alloy_primitives::B256;
     use alloy_rpc_types_eth::BlockNumHash;
@@ -165,7 +168,7 @@
     }
 
     #[async_trait]
-    impl ManagedNodeApiProvider for MockNode {
+    impl ManagedNodeDataProvider for MockNode {
         async fn output_v0_at_timestamp(
             &self,
             _timestamp: u64,
@@ -186,7 +189,10 @@
         ) -> Result<BlockInfo, ManagedNodeError> {
             Ok(BlockInfo::default())
         }
+    }
 
+    #[async_trait]
+    impl ManagedNodeController for MockNode {
         async fn update_finalized(
             &self,
             _finalized_block_id: BlockNumHash,
@@ -208,6 +214,10 @@
         ) -> Result<(), ManagedNodeError> {
             Ok(())
         }
+
+        async fn reset(&self) -> Result<(), ManagedNodeError> {
+            Ok(())
+        }
     }
 
     mock!(

crates/supervisor/core/src/chain_processor/task.rs

@@ -3,7 +3,9 @@
 use alloy_primitives::ChainId;
 use kona_interop::{BlockReplacement, DerivedRefPair};
 use kona_protocol::BlockInfo;
-use kona_supervisor_storage::{DerivationStorageWriter, HeadRefStorageWriter, LogStorageWriter};
+use kona_supervisor_storage::{
+    DerivationStorageWriter, HeadRefStorageWriter, LogStorageWriter, StorageError,
+};
 use std::{fmt::Debug, sync::Arc};
 use tokio::sync::mpsc;
 use tokio_util::sync::CancellationToken;
@@ -175,7 +177,7 @@
                     .await;
             }
             ChainEvent::DerivationOriginUpdate { origin } => {
-                let _ = self.handle_derivation_origin_update(origin).inspect_err(|err| {
+                let _ = self.handle_derivation_origin_update(origin).await.inspect_err(|err| {
                     error!(
                         target: "chain_processor",
                         chain_id = self.chain_id,
@@ -270,7 +272,7 @@
         Ok(finalized_derived_block)
     }
 
-    fn handle_derivation_origin_update(
+    async fn handle_derivation_origin_update(
         &self,
         origin: BlockInfo,
     ) -> Result<(), ChainProcessorError> {
@@ -280,9 +282,31 @@
             block_number = origin.number,
             "Processing derivation origin update"
         );
-        self.state_manager.update_current_l1(origin)?;
-        self.state_manager.save_source_block(origin)?;
-        Ok(())
+        match self.state_manager.save_source_block(origin) {
+            Ok(_) => {
+                self.state_manager.update_current_l1(origin)?;
+                Ok(())
+            }
+            Err(StorageError::BlockOutOfOrder) => {
+                error!(
+                    target: "chain_processor",
+                    chain_id = self.chain_id,
+                    block_number = origin.number,
+                    "Source block out of order detected, resetting managed node"
+                );
+
+                if let Err(err) = self.managed_node.reset().await {
+                    error!(
+                        target: "chain_processor",
+                        chain_id = self.chain_id,
+                        %err,
+                        "Failed to reset managed node after block out of order"
+                    );
+                }
+                Err(StorageError::BlockOutOfOrder.into())
+            }
+            Err(err) => Err(err.into()),
+        }
     }
 
     async fn handle_safe_event(
@@ -295,8 +319,37 @@
             block_number = derived_ref_pair.derived.number,
             "Processing local safe derived block pair"
         );
-        self.state_manager.save_derived_block(derived_ref_pair)?;
-        Ok(derived_ref_pair.derived)
+        match self.state_manager.save_derived_block(derived_ref_pair) {
+            Ok(_) => Ok(derived_ref_pair.derived),
+            Err(StorageError::BlockOutOfOrder) => {
+                error!(
+                    target: "chain_processor",
+                    chain_id = self.chain_id,
+                    block_number = derived_ref_pair.derived.number,
+                    "Block out of order detected, resetting managed node"
+                );
+
+                if let Err(err) = self.managed_node.reset().await {
+                    error!(
+                        target: "chain_processor",
+                        chain_id = self.chain_id,
+                        %err,
+                        "Failed to reset managed node after block out of order"
+                    );
+                }
+                Err(StorageError::BlockOutOfOrder.into())
+            }
+            Err(err) => {
+                error!(
+                    target: "chain_processor",
+                    chain_id = self.chain_id,
+                    block_number = derived_ref_pair.derived.number,
+                    %err,
+                    "Failed to save derived block pair"
+                );
+                Err(err.into())
+            }
+        }
     }
 
     async fn handle_unsafe_event(
@@ -352,7 +405,10 @@
     use super::*;
     use crate::{
         event::ChainEvent,
-        syncnode::{ManagedNodeApiProvider, ManagedNodeError, NodeSubscriber, ReceiptProvider},
+        syncnode::{
+            ManagedNodeController, ManagedNodeDataProvider, ManagedNodeError, NodeSubscriber,
+            ReceiptProvider,
+        },
     };
     use alloy_primitives::B256;
     use alloy_rpc_types_eth::BlockNumHash;
@@ -377,46 +433,51 @@
                 &self,
                 _event_tx: mpsc::Sender<ChainEvent>,
             ) -> Result<(), ManagedNodeError>;
-    }
+        }
 
-    #[async_trait]
-    impl ReceiptProvider for Node {
-        async fn fetch_receipts(&self, _block_hash: B256) -> Result<Receipts, ManagedNodeError>;
-    }
+        #[async_trait]
+        impl ReceiptProvider for Node {
+            async fn fetch_receipts(&self, _block_hash: B256) -> Result<Receipts, ManagedNodeError>;
+        }
 
-    #[async_trait]
-    impl ManagedNodeApiProvider for Node {
-        async fn output_v0_at_timestamp(
-            &self,
-            _timestamp: u64,
-        ) -> Result<OutputV0, ManagedNodeError>;
-
-        async fn pending_output_v0_at_timestamp(
-            &self,
-            _timestamp: u64,
-        ) -> Result<OutputV0, ManagedNodeError>;
-
-        async fn l2_block_ref_by_timestamp(
-            &self,
-            _timestamp: u64,
-        ) -> Result<BlockInfo, ManagedNodeError>;
-
-        async fn update_finalized(
-            &self,
-            _finalized_block_id: BlockNumHash,
-        ) -> Result<(), ManagedNodeError>;
-
-        async fn update_cross_unsafe(
-            &self,
-            cross_unsafe_block_id: BlockNumHash,
-        ) -> Result<(), ManagedNodeError>;
-
-        async fn update_cross_safe(
-            &self,
-            source_block_id: BlockNumHash,
-            derived_block_id: BlockNumHash,
-        ) -> Result<(), ManagedNodeError>;
-    }
+        #[async_trait]
+        impl ManagedNodeDataProvider for Node {
+            async fn output_v0_at_timestamp(
+                &self,
+                _timestamp: u64,
+            ) -> Result<OutputV0, ManagedNodeError>;
+
+            async fn pending_output_v0_at_timestamp(
+                &self,
+                _timestamp: u64,
+            ) -> Result<OutputV0, ManagedNodeError>;
+
+            async fn l2_block_ref_by_timestamp(
+                &self,
+                _timestamp: u64,
+            ) -> Result<BlockInfo, ManagedNodeError>;
+        }
+
+        #[async_trait]
+        impl ManagedNodeController for Node {
+            async fn update_finalized(
+                &self,
+                _finalized_block_id: BlockNumHash,
+            ) -> Result<(), ManagedNodeError>;
+
+            async fn update_cross_unsafe(
+                &self,
+                cross_unsafe_block_id: BlockNumHash,
+            ) -> Result<(), ManagedNodeError>;
+
+            async fn update_cross_safe(
+                &self,
+                source_block_id: BlockNumHash,
+                derived_block_id: BlockNumHash,
+            ) -> Result<(), ManagedNodeError>;
+
+            async fn reset(&self) -> Result<(), ManagedNodeError>;
+        }
     );
 
     mock!(

crates/supervisor/core/src/supervisor.rs

@@ -28,7 +28,7 @@ use crate::{
     event::ChainEvent,
     l1_watcher::L1Watcher,
     safety_checker::{CrossSafePromoter, CrossUnsafePromoter},
-    syncnode::{Client, ManagedNode, ManagedNodeApiProvider, ManagedNodeClient},
+    syncnode::{Client, ManagedNode, ManagedNodeClient, ManagedNodeDataProvider},
 };
 
 /// Defines the service for the Supervisor core logic.

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

@@ -21,9 +21,9 @@ pub enum ManagedNodeError {
     #[error("failed to authenticate: {0}")]
     Authentication(#[from] AuthenticationError),
 
-    /// Database provider is not set for the managed node.
-    #[error("database not initialised for managed node")]
-    DatabaseNotInitialised,
+    /// Represents an error that occurred while fetching data from the storage.
+    #[error(transparent)]
+    StorageError(#[from] StorageError),
 }
 
 impl PartialEq for ManagedNodeError {

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

@@ -8,7 +8,10 @@ mod error;
 pub use error::{AuthenticationError, ManagedEventTaskError, ManagedNodeError, SubscriptionError};
 
 mod traits;
-pub use traits::{ManagedNodeApiProvider, ManagedNodeProvider, NodeSubscriber, ReceiptProvider};
+pub use traits::{
+    ManagedNodeController, ManagedNodeDataProvider, ManagedNodeProvider, NodeSubscriber,
+    ReceiptProvider,
+};
 
 mod client;
 pub use client::{Client, ClientConfig, ManagedNodeClient};

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

@@ -17,8 +17,8 @@
 use tracing::{error, info, warn};
 
 use super::{
-    ManagedNodeApiProvider, ManagedNodeClient, ManagedNodeError, NodeSubscriber, ReceiptProvider,
-    SubscriptionError, resetter::Resetter, task::ManagedEventTask,
+    ManagedNodeClient, ManagedNodeController, ManagedNodeDataProvider, ManagedNodeError,
+    NodeSubscriber, ReceiptProvider, SubscriptionError, resetter::Resetter, task::ManagedEventTask,
 };
 use crate::event::ChainEvent;
 
@@ -29,8 +29,6 @@
 pub struct ManagedNode<DB, C> {
     /// The attached web socket client
     client: Arc<C>,
-    /// The database provider for fetching information
-    db_provider: Arc<DB>,
     /// Shared L1 provider for fetching receipts
     l1_provider: RootProvider<Ethereum>,
     /// Resetter for handling node resets
@@ -53,16 +51,9 @@
         cancel_token: CancellationToken,
         l1_provider: RootProvider<Ethereum>,
     ) -> Self {
-        let resetter = Arc::new(Resetter::new(client.clone(), db_provider.clone()));
-
-        Self {
-            client,
-            db_provider,
-            resetter,
-            cancel_token,
-            task_handle: Mutex::new(None),
-            l1_provider,
-        }
+        let resetter = Arc::new(Resetter::new(client.clone(), db_provider));
+
+        Self { client, resetter, cancel_token, task_handle: Mutex::new(None), l1_provider }
     }
 
     /// Returns the [`ChainId`] of the [`ManagedNode`].
@@ -106,7 +97,6 @@
         let task = ManagedEventTask::new(
             self.client.clone(),
             self.l1_provider.clone(),
-            self.db_provider.clone(),
             self.resetter.clone(),
             event_tx,
         );
@@ -178,7 +168,7 @@
 }
 
 #[async_trait]
-impl<DB, C> ManagedNodeApiProvider for ManagedNode<DB, C>
+impl<DB, C> ManagedNodeDataProvider for ManagedNode<DB, C>
 where
     DB: LogStorageReader + DerivationStorageReader + HeadRefStorageReader + Send + Sync + 'static,
     C: ManagedNodeClient + Send + Sync + 'static,
@@ -200,7 +190,14 @@
     ) -> Result<BlockInfo, ManagedNodeError> {
         self.client.l2_block_ref_by_timestamp(timestamp).await
     }
+}
 
+#[async_trait]
+impl<DB, C> ManagedNodeController for ManagedNode<DB, C>
+where
+    DB: LogStorageReader + DerivationStorageReader + HeadRefStorageReader + Send + Sync + 'static,
+    C: ManagedNodeClient + Send + Sync + 'static,
+{
     async fn update_finalized(
         &self,
         finalized_block_id: BlockNumHash,
@@ -222,4 +219,8 @@
     ) -> Result<(), ManagedNodeError> {
         self.client.update_cross_safe(source_block_id, derived_block_id).await
     }
+
+    async fn reset(&self) -> Result<(), ManagedNodeError> {
+        self.resetter.reset().await
+    }
 }