Periodic L1 synchronization

kairos-server/.env

@@ -1,2 +1,3 @@
 KAIROS_SERVER_SOCKET_ADDR="127.0.0.1:7893"
 KAIROS_SERVER_CASPER_RPC="http://127.0.0.1:11101/rpc"
+KAIROS_SERVER_CASPER_CONTRACT_HASH="0000000000000000000000000000000000000000000000000000000000000000"

kairos-server/Cargo.toml

@@ -39,6 +39,8 @@ kairos-circuit-logic = { path = "../kairos-prover/kairos-circuit-logic", feature
 kairos-trie = { git = "https://github.com/cspr-rad/kairos-trie" }
 sha2 = "0.10"
 reqwest = "0.12"
+casper-event-toolkit = { git = "https://github.com/koxu1996/casper-event-toolkit.git", version = "0.1.3" }
+thiserror = "1.0"
 
 [dev-dependencies]
 proptest = "1"

kairos-server/src/config.rs

@@ -6,15 +6,18 @@ use std::{fmt, str::FromStr};
 pub struct ServerConfig {
     pub socket_addr: SocketAddr,
     pub casper_rpc: Url,
+    pub casper_contract_hash: String,
 }
 
 impl ServerConfig {
     pub fn from_env() -> Result<Self, String> {
         let socket_addr = parse_env_as::<SocketAddr>("KAIROS_SERVER_SOCKET_ADDR")?;
         let casper_rpc = parse_env_as::<Url>("KAIROS_SERVER_CASPER_RPC")?;
+        let casper_contract_hash = parse_env_as::<String>("KAIROS_SERVER_CASPER_CONTRACT_HASH")?;
         Ok(Self {
             socket_addr,
             casper_rpc,
+            casper_contract_hash,
         })
     }
 }

kairos-server/src/l1_sync/error.rs

@@ -0,0 +1,23 @@
+use thiserror::Error;
+
+#[derive(Error, Debug)]
+pub enum L1SyncError {
+    /// Casper Event Toolkit error.
+    #[error("toolkit error: {error}")]
+    ToolkitError {
+        #[from]
+        error: casper_event_toolkit::error::ToolkitError,
+    },
+
+    /// Communication error.
+    #[error("channel error: {0}")]
+    BrokenChannel(String),
+
+    /// Initialization error.
+    #[error("Initialization error: {0}")]
+    InitializationError(String),
+
+    /// Error that we cannot recover from.
+    #[error("Unexpected error: {0}")]
+    UnexpectedError(String),
+}

kairos-server/src/l1_sync/event_manager.rs

@@ -0,0 +1,89 @@
+use std::sync::Arc;
+
+use casper_event_toolkit::fetcher::{Fetcher, Schemas};
+use casper_event_toolkit::metadata::CesMetadataRef;
+use casper_event_toolkit::rpc::client::CasperClient;
+
+use crate::state::ServerStateInner;
+use kairos_circuit_logic::transactions::{KairosTransaction, L1Deposit};
+
+use super::error::L1SyncError;
+
+pub struct EventManager {
+    next_event_id: u32,
+    fetcher: Option<Fetcher>,
+    schemas: Option<Schemas>,
+    server_state: Arc<ServerStateInner>,
+}
+
+impl EventManager {
+    pub fn new(server_state: Arc<ServerStateInner>) -> Self {
+        EventManager {
+            next_event_id: 0,
+            fetcher: None,
+            schemas: None,
+            server_state,
+        }
+    }
+
+    /// Initializes state by building CES fetcher and obtaining schemas.
+    pub async fn initialize(
+        &mut self,
+        rpc_url: &str,
+        contract_hash: &str,
+    ) -> Result<(), L1SyncError> {
+        tracing::info!("Initializing event manager");
+
+        let client = CasperClient::new(rpc_url);
+        let metadata = CesMetadataRef::fetch_metadata(&client, contract_hash).await?;
+        tracing::debug!("Metadata fetched successfully");
+
+        let fetcher = Fetcher {
+            client,
+            ces_metadata: metadata,
+        };
+        let schemas = fetcher.fetch_schema().await?;
+        tracing::debug!("Schemas fetched successfully");
+
+        self.fetcher = Some(fetcher);
+        self.schemas = Some(schemas);
+
+        Ok(())
+    }
+
+    /// Processes new events starting from the last known event ID.
+    pub async fn process_new_events(&mut self) -> Result<(), L1SyncError> {
+        tracing::info!("Looking for new events");
+
+        // Ensure fetcher and schemas are initialized
+        let fetcher = self.fetcher.as_ref().ok_or_else(|| {
+            L1SyncError::InitializationError("Fetcher not initialized".to_string())
+        })?;
+        let schemas = self.schemas.as_ref().ok_or_else(|| {
+            L1SyncError::InitializationError("Schemas not initialized".to_string())
+        })?;
+
+        let num_events = fetcher.fetch_events_count().await?;
+        for i in self.next_event_id..num_events {
+            let event = fetcher.fetch_event(i, schemas).await?;
+            tracing::debug!("Event {} fetched: {:?}.", i, event);
+
+            // TODO: Parse full transaction data from event, then push it to Data Availability layer.
+
+            // TODO: Once we have ASN transaction, it should be converted and pushed into batch.
+            let recipient: Vec<u8> = "cafebabe".into();
+            let txn = KairosTransaction::Deposit(L1Deposit {
+                amount: 100,
+                recipient,
+            });
+            self.server_state
+                .batch_state_manager
+                .enqueue_transaction(txn)
+                .await
+                .map_err(|e| L1SyncError::UnexpectedError(format!("unable to batch tx: {}", e)))?;
+            self.next_event_id = i + 1;
+        }
+
+        Ok(())
+    }
+}

kairos-server/src/l1_sync/interval_trigger.rs

@@ -0,0 +1,18 @@
+use tokio::time::{self, Duration};
+
+use std::sync::Arc;
+
+use super::service::L1SyncService;
+
+pub async fn run(sync_service: Arc<L1SyncService>) {
+    let mut interval = time::interval(Duration::from_secs(30));
+
+    loop {
+        interval.tick().await;
+
+        tracing::debug!("Triggering periodic L1 sync");
+        let _ = sync_service.trigger_sync().await.map_err(|e| {
+            tracing::error!("Unable to trigger sync: {}", e);
+        });
+    }
+}

kairos-server/src/l1_sync/mod.rs

@@ -0,0 +1,5 @@
+pub mod error;
+pub mod event_manager;
+pub mod service;
+
+pub mod interval_trigger;

kairos-server/src/l1_sync/service.rs

@@ -0,0 +1,94 @@
+use crate::state::ServerStateInner;
+
+use super::error::L1SyncError;
+use super::event_manager::EventManager;
+
+use tokio::sync::mpsc;
+use tokio::sync::oneshot;
+
+use std::sync::Arc;
+
+pub enum SyncCommand {
+    TriggerSync(oneshot::Sender<()>),
+    // NOTE: More commands can be here.
+}
+
+pub struct L1SyncService {
+    command_sender: Option<mpsc::Sender<SyncCommand>>,
+    //event_manager_handle: Option<tokio::task::JoinHandle<()>>,
+    server_state: Arc<ServerStateInner>,
+}
+
+impl L1SyncService {
+    pub fn new(server_state: Arc<ServerStateInner>) -> Self {
+        L1SyncService {
+            command_sender: None,
+            //event_manager_handle: None,
+            server_state,
+        }
+    }
+
+    pub async fn initialize(
+        &mut self,
+        rpc_url: String,
+        contract_hash: String,
+    ) -> Result<(), L1SyncError> {
+        let mut event_manager = EventManager::new(self.server_state.clone());
+        event_manager.initialize(&rpc_url, &contract_hash).await?;
+
+        let (tx, rx) = mpsc::channel(32);
+        self.command_sender = Some(tx);
+        let _handle = tokio::spawn(async move {
+            run_event_manager(rx, event_manager).await;
+        });
+        //self.event_manager_handle = Some(handle);
+
+        Ok(())
+    }
+
+    pub async fn trigger_sync(&self) -> Result<(), L1SyncError> {
+        let command_sender = self.command_sender.as_ref().ok_or_else(|| {
+            L1SyncError::InitializationError("Command sender not available".to_string())
+        })?;
+
+        let (tx, rx) = oneshot::channel();
+        command_sender
+            .send(SyncCommand::TriggerSync(tx))
+            .await
+            .map_err(|e| L1SyncError::BrokenChannel(format!("Unable to send trigger: {}", e)))?;
+        rx.await.map_err(|e| {
+            L1SyncError::BrokenChannel(format!("Unable to receive trigger ack: {}", e))
+        })?;
+
+        Ok(())
+    }
+}
+
+/// Handles incoming commands and delegates tasks to EventManager.
+async fn run_event_manager(mut rx: mpsc::Receiver<SyncCommand>, mut event_manager: EventManager) {
+    tracing::debug!("Event manager running and waiting for commands");
+    while let Some(command) = rx.recv().await {
+        let _ = handle_command(command, &mut event_manager)
+            .await
+            .map_err(|e| match e {
+                L1SyncError::UnexpectedError(e) => panic!("Unrecoverable error: {}", e),
+                _ => tracing::error!("Transient error: {}", e),
+            });
+    }
+}
+
+async fn handle_command(
+    command: SyncCommand,
+    event_manager: &mut EventManager,
+) -> Result<(), L1SyncError> {
+    match command {
+        SyncCommand::TriggerSync(completion_ack) => {
+            event_manager.process_new_events().await?;
+            completion_ack
+                .send(())
+                .map_err(|_| L1SyncError::BrokenChannel("Sender dropped".to_string()))?;
+        }
+    }
+
+    Ok(())
+}

kairos-server/src/lib.rs

@@ -3,6 +3,7 @@ pub mod errors;
 pub mod routes;
 pub mod state;
 
+mod l1_sync;
 mod utils;
 
 use std::sync::Arc;
@@ -13,6 +14,7 @@ use axum_extra::routing::RouterExt;
 pub use errors::AppErr;
 
 use crate::config::ServerConfig;
+use crate::l1_sync::service::L1SyncService;
 use crate::state::{BatchStateManager, ServerState, ServerStateInner};
 
 /// TODO: support secp256k1
@@ -38,6 +40,33 @@ pub fn app_router(state: ServerState) -> Router {
         .with_state(state)
 }
 
+pub async fn run_l1_sync(config: ServerConfig, server_state: Arc<ServerStateInner>) {
+    // Make sure real contract hash was provided.
+    if config.casper_contract_hash
+        == "0000000000000000000000000000000000000000000000000000000000000000"
+    {
+        tracing::warn!(
+            "Casper contract hash not configured, L1 synchronization will NOT be enabled."
+        );
+        return;
+    }
+
+    // Initialize L1 synchronizer.
+    let mut l1_sync_service = L1SyncService::new(server_state);
+    let _ = l1_sync_service
+        .initialize(config.casper_rpc.to_string(), config.casper_contract_hash)
+        .await
+        .map_err(|e| {
+            panic!("Event manager failed to initialize: {}", e);
+        });
+
+    // Run periodic synchronization.
+    // TODO: Add additional SSE trigger.
+    tokio::spawn(async move {
+        l1_sync::interval_trigger::run(l1_sync_service.into()).await;
+    });
+}
+
 pub async fn run(config: ServerConfig) {
     let listener = tokio::net::TcpListener::bind(config.socket_addr)
         .await
@@ -48,6 +77,9 @@ pub async fn run(config: ServerConfig) {
         batch_state_manager: BatchStateManager::new_empty(),
         server_config: config.clone(),
     });
+
+    run_l1_sync(config.clone(), state.clone()).await;
+
     let app = app_router(state);
 
     axum::serve(listener, app)

kairos-server/tests/transactions.rs

@@ -49,6 +49,8 @@ fn new_test_app_with_casper_node(casper_node_url: &Url) -> TestServer {
         server_config: ServerConfig {
             socket_addr: "0.0.0.0:0".parse().unwrap(),
             casper_rpc: casper_node_url.clone(),
+            casper_contract_hash:
+                "0000000000000000000000000000000000000000000000000000000000000000".to_string(),
         },
     });