feat(supervisor): integrated cross safety checker job

crates/supervisor/core/src/error.rs

@@ -1,6 +1,6 @@
 //! [`SupervisorService`](crate::SupervisorService) errors.
 
-use crate::{ChainProcessorError, syncnode::ManagedNodeError};
+use crate::{ChainProcessorError, CrossSafetyError, syncnode::ManagedNodeError};
 use jsonrpsee::types::{ErrorCode, ErrorObjectOwned};
 use kona_supervisor_storage::StorageError;
 use kona_supervisor_types::AccessListError;
@@ -41,6 +41,10 @@ pub enum SupervisorError {
     /// Indicates the error occurred while parsing the access_list
     #[error(transparent)]
     AccessListError(#[from] AccessListError),
+
+    /// Indicated the error occurred while initializing the safety checker jobs
+    #[error(transparent)]
+    CrossSafetyCheckerError(#[from] CrossSafetyError),
 }
 
 impl From<SupervisorError> for ErrorObjectOwned {
@@ -53,6 +57,7 @@ impl From<SupervisorError> for ErrorObjectOwned {
             SupervisorError::StorageError(_) |
             SupervisorError::ManagedNodeError(_) |
             SupervisorError::ChainProcessorError(_) |
+            SupervisorError::CrossSafetyCheckerError(_) |
             SupervisorError::AccessListError(_) => ErrorObjectOwned::from(ErrorCode::InternalError),
             SupervisorError::DataAvailability(err) => err.into(),
         }

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

@@ -4,7 +4,7 @@ use op_alloy_consensus::interop::SafetyLevel;
 use thiserror::Error;
 
 /// Errors returned when validating cross-chain message dependencies.
-#[derive(Debug, Error)]
+#[derive(Debug, Error, Eq, PartialEq)]
 pub enum CrossSafetyError {
     /// Indicates a failure while accessing storage during dependency checking.
     #[error("storage error: {0}")]

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

@@ -6,7 +6,7 @@
 use kona_supervisor_storage::CrossChainSafetyProvider;
 use op_alloy_consensus::interop::SafetyLevel;
 use tokio_util::sync::CancellationToken;
-use tracing::{debug, info, warn};
+use tracing::{info, warn};
 
 /// A background job that promotes blocks to a target safety level on a given chain.
 ///
@@ -75,7 +75,7 @@
                 _ = async {
                     match self.promote_next_block(&checker) {
                         Ok(block_info) => {
-                            debug!(
+                            info!(
                                 target: "safety_checker",
                                 chain_id = self.chain_id,
                                 target_level = %self.target_level,
@@ -84,13 +84,19 @@
                             );
                         }
                         Err(err) => {
-                            warn!(
-                                target: "safety_checker",
-                                chain_id = self.chain_id,
-                                target_level = %self.target_level,
-                                %err,
-                                "Error promoting next candidate block"
-                            );
+                            match err {
+                                 // don't spam warnings if head is already on top - nothing to promote
+                                CrossSafetyError::NoBlockToPromote => {},
+                                _ => {
+                                    warn!(
+                                        target: "safety_checker",
+                                        chain_id = self.chain_id,
+                                        target_level = %self.target_level,
+                                        %err,
+                                        "Error promoting next candidate block"
+                                    );
+                                }
+                            }
                             tokio::time::sleep(self.interval).await;
                         }
                     }

crates/supervisor/core/src/supervisor.rs

@@ -17,12 +17,12 @@
 use kona_supervisor_types::{SuperHead, parse_access_list};
 use op_alloy_rpc_types::SuperchainDAError;
 use reqwest::Url;
-use std::{collections::HashMap, sync::Arc};
+use std::{collections::HashMap, sync::Arc, time::Duration};
 use tokio_util::sync::CancellationToken;
 use tracing::{error, info, warn};
 
 use crate::{
-    ChainProcessor, SupervisorError,
+    ChainProcessor, CrossSafetyCheckerJob, SupervisorError,
     config::Config,
     l1_watcher::L1Watcher,
     syncnode::{ManagedNode, ManagedNodeApiProvider},
@@ -131,6 +131,7 @@
         self.init_managed_nodes().await?;
         self.init_chain_processor().await?;
         self.init_l1_watcher()?;
+        self.init_cross_safety_checker().await?;
         Ok(())
     }
 
@@ -166,6 +167,38 @@
         }
         Ok(())
     }
+    async fn init_cross_safety_checker(&self) -> Result<(), SupervisorError> {
+        for (&chain_id, config) in &self.config.rollup_config_set.rollups {
+            let db = Arc::clone(&self.database_factory);
+            let cancel = self.cancel_token.clone();
+
+            let cross_safe_job = CrossSafetyCheckerJob::new(
+                chain_id,
+                db.clone(),
+                cancel.clone(),
+                Duration::from_secs(config.block_time),
+                SafetyLevel::CrossSafe,
+            )?;
+
+            tokio::spawn(async move {
+                cross_safe_job.run().await;
+            });
+
+            let cross_unsafe_job = CrossSafetyCheckerJob::new(
+                chain_id,
+                db,
+                cancel,
+                Duration::from_secs(config.block_time),
+                SafetyLevel::CrossUnsafe,
+            )?;
+
+            tokio::spawn(async move {
+                cross_unsafe_job.run().await;
+            });
+        }
+
+        Ok(())
+    }
 
     async fn init_managed_nodes(&mut self) -> Result<(), SupervisorError> {
         for config in self.config.l2_consensus_nodes_config.iter() {