feat(supervisor/core): Handle unsafe reorg

crates/supervisor/core/src/lib.rs

@@ -25,3 +25,6 @@ pub mod syncnode;
 
 pub mod safety_checker;
 pub use safety_checker::{CrossSafetyCheckerJob, CrossSafetyError};
+
+mod rewinder;
+pub use rewinder::{ChainRewinder, ChainRewinderError};

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

@@ -0,0 +1,100 @@
+use alloy_primitives::ChainId;
+use derive_more::Constructor;
+use kona_interop::DerivedRefPair;
+use kona_supervisor_storage::{LogStorageReader, StorageError, StorageRewinder};
+use thiserror::Error;
+use tracing::{error, info, warn};
+
+/// Initiates supervisor-level rewinds based on chain events or storage conflicts.
+///
+/// This coordinates per-chain rewind logic using the underlying [`StorageRewinder`] implementation,
+/// and encapsulates the context in which rewinds should occur.
+///
+/// It is used in response to:
+/// - Local derivation conflicts (failure updating supervisor state)
+/// - L1-originated reorgs affecting derived state
+#[derive(Debug, Constructor)]
+pub struct ChainRewinder<DB> {
+    chain_id: ChainId,
+    db: DB,
+}
+
+#[expect(dead_code)] // todo: to be removed in the follow up PR
+impl<DB> ChainRewinder<DB>
+where
+    DB: StorageRewinder + LogStorageReader,
+{
+    /// Handles a local reorg by rewinding supervisor state from the conflicting derived pair.
+    ///
+    /// This is triggered when an update to supervisor storage fails due to an
+    /// integrity violation (e.g., mismatched on storing local safe block hash).
+    fn handle_local_reorg(&self, derived_pair: &DerivedRefPair) -> Result<(), StorageError> {
+        warn!(
+            target: "rewinder",
+            chain = %self.chain_id,
+            derived_block = %derived_pair.derived,
+            "Local derivation conflict detected — rewinding..."
+        );
+        // get the same block from log storage
+        let conflicting_block =
+            self.db.get_block(derived_pair.derived.number).inspect_err(|err| {
+                error!(
+                    target: "rewinder",
+                    chain = %self.chain_id,
+                    block_number = derived_pair.derived.number,
+                    %err,
+                    "Error retrieving conflicting block for reorg"
+                );
+            })?;
+
+        // cross-check whether the block is conflicting
+        if conflicting_block == derived_pair.derived {
+            return Ok(())
+        }
+
+        // rewind the log storage to remove all the blocks till the conflicting one
+        self.db.rewind_log_storage(&conflicting_block.id()).inspect_err(|err| {
+            error!(
+                target: "rewinder",
+                chain = %self.chain_id,
+                block_number = derived_pair.derived.number,
+                %err,
+                "Error rewinding the log storage"
+            );
+        })?;
+
+        // todo: sync the log storage - to prevent a reset
+        // todo: save the derived_pair - now it should succeed
+
+        info!(
+            target: "rewinder",
+            chain = self.chain_id,
+            "Rewind successful after local derivation conflict"
+        );
+
+        Ok(())
+    }
+
+    /// Handles a rewind due to an L1 reorg.
+    ///
+    /// This method is expected to revert supervisor state based on the L1 reorg by finding the new
+    /// valid state and removing any derived data that is no longer valid due to upstream
+    /// reorganization.
+    fn handle_l1_reorg(&self) -> Result<(), StorageError> {
+        warn!(
+            target: "rewinder",
+            chain = self.chain_id,
+            "L1 reorg handling is not yet implemented. Skipping rewind."
+        );
+
+        Ok(())
+    }
+}
+
+/// Error type for the [`ChainRewinder`].
+#[derive(Error, Debug, PartialEq, Eq)]
+pub enum ChainRewinderError {
+    /// Failed on storage operations
+    #[error(transparent)]
+    StorageError(#[from] StorageError),
+}

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

@@ -0,0 +1,5 @@
+//! Rewinder module for reverting supervisor state during re-org
+
+mod chain;
+
+pub use chain::{ChainRewinder, ChainRewinderError};

crates/supervisor/storage/src/chaindb.rs

@@ -1,7 +1,7 @@
 //! Main database access structure and transaction contexts.
 
 use crate::{
-    Metrics, Rewinder,
+    Metrics, StorageRewinder,
     error::StorageError,
     providers::{DerivationProvider, LogProvider, SafetyHeadRefProvider},
     traits::{
@@ -380,7 +380,7 @@ impl HeadRefStorageWriter for ChainDb {
     }
 }
 
-impl Rewinder for ChainDb {
+impl StorageRewinder for ChainDb {
     fn rewind_log_storage(&self, to: &BlockNumHash) -> Result<(), StorageError> {
         self.observe_call("rewind_log_storage", || {
             self.env.update(|tx| {

crates/supervisor/storage/src/lib.rs

@@ -40,5 +40,5 @@ mod traits;
 pub use traits::{
     CrossChainSafetyProvider, DerivationStorage, DerivationStorageReader, DerivationStorageWriter,
     FinalizedL1Storage, HeadRefStorage, HeadRefStorageReader, HeadRefStorageWriter, LogStorage,
-    LogStorageReader, LogStorageWriter, Rewinder,
+    LogStorageReader, LogStorageWriter, StorageRewinder,
 };

crates/supervisor/storage/src/traits.rs

@@ -392,7 +392,7 @@ pub trait CrossChainSafetyProvider {
 /// and safety head references from the latest block back to a specified block number (inclusive).
 /// It is typically used during chain reorganizations or when invalid blocks are detected and need
 /// to be rolled back.
-pub trait Rewinder {
+pub trait StorageRewinder {
     /// Rewinds the log storage from the latest block down to the specified block (inclusive).
     ///
     /// # Arguments