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

@@ -30,6 +30,8 @@ kairos-tx = { path = "../kairos-tx" }
 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.1" }
+thiserror = "1.0"
 
 [dev-dependencies]
 proptest = "1"

kairos-server/src/config.rs

@@ -3,19 +3,22 @@ use std::{fmt, str::FromStr};
 
 use reqwest::Url;
 
-#[derive(Debug)]
+#[derive(Clone, Debug)]
 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,19 @@
+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),
+
+    /// Error that we cannot recover from.
+    #[error("Unexpected error: {0}")]
+    UnexpectedError(String),
+}

kairos-server/src/l1_sync/event_manager.rs

@@ -0,0 +1,85 @@
+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::transactions::{Deposit, Signed, Transaction};
+use crate::state::BatchStateManager;
+
+use super::error::L1SyncError;
+
+pub struct EventManager {
+    next_event_id: u32,
+    fetcher: Option<Fetcher>,
+    schemas: Option<Schemas>,
+    batch_service: Arc<BatchStateManager>,
+}
+
+impl EventManager {
+    pub fn new(batch_service: Arc<BatchStateManager>) -> Self {
+        EventManager {
+            next_event_id: 0,
+            fetcher: None,
+            schemas: None,
+            batch_service,
+        }
+    }
+
+    /// 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");
+
+        if let Some(fetcher) = &self.fetcher {
+            let schemas = self.schemas.as_ref().unwrap(); // Assuming schemas are already loaded
+            let num_events = fetcher.fetch_events_count().await?;
+            for i in self.next_event_id..num_events {
+                let event = fetcher.fetch_event(i.into(), 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 txn = Signed {
+                    public_key: "cafebabe".into(),
+                    nonce: 0,
+                    transaction: Transaction::Deposit(Deposit { amount: 100 }),
+                };
+                self.batch_service
+                    .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,20 @@
+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 result = sync_service.trigger_sync().await;
+
+        if let Err(e) = result {
+            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,102 @@
+use crate::state::BatchStateManager;
+
+use super::error::L1SyncError;
+use super::event_manager::EventManager;
+
+use tokio::sync::mpsc;
+use tokio::sync::oneshot;
+use tokio::sync::Mutex;
+
+use std::sync::Arc;
+
+pub enum SyncCommand {
+    Initialize(String, String, oneshot::Sender<()>),
+    TriggerSync(oneshot::Sender<()>),
+}
+
+pub struct L1SyncService {
+    command_sender: mpsc::Sender<SyncCommand>,
+    //event_manager: Arc<Mutex<EventManager>>, // NOTE: It could be stored for shared access.
+}
+
+impl L1SyncService {
+    pub async fn new(batch_service: Arc<BatchStateManager>) -> Self {
+        let (tx, rx) = mpsc::channel(32);
+        let event_manager = Arc::new(Mutex::new(EventManager::new(batch_service.clone())));
+        let event_manager_clone = event_manager.clone();
+
+        tokio::spawn(async move {
+            run_event_manager(rx, event_manager_clone).await;
+        });
+
+        L1SyncService {
+            command_sender: tx,
+            //event_manager,
+        }
+    }
+
+    pub async fn initialize(
+        &self,
+        rpc_url: String,
+        contract_hash: String,
+    ) -> Result<(), L1SyncError> {
+        let (tx, rx) = oneshot::channel();
+        self.command_sender
+            .send(SyncCommand::Initialize(rpc_url, contract_hash, tx))
+            .await
+            .map_err(|e| L1SyncError::BrokenChannel(format!("Unable to send initialize: {}", e)))?;
+        rx.await.map_err(|e| {
+            L1SyncError::BrokenChannel(format!("Unable to receive initialize ack: {}", e))
+        })?;
+
+        Ok(())
+    }
+
+    pub async fn trigger_sync(&self) -> Result<(), L1SyncError> {
+        let (tx, rx) = oneshot::channel();
+        self.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>,
+    event_manager: Arc<Mutex<EventManager>>,
+) {
+    tracing::debug!("Event manager running and waiting for commands");
+    while let Some(command) = rx.recv().await {
+        match handle_command(command, event_manager.clone()).await {
+            Ok(()) => {}
+            Err(L1SyncError::UnexpectedError(e)) => panic!("Unrecoverable error: {}", e),
+            Err(e) => tracing::error!("Transient error: {}", e),
+        }
+    }
+}
+
+async fn handle_command(
+    command: SyncCommand,
+    event_manager: Arc<Mutex<EventManager>>,
+) -> Result<(), L1SyncError> {
+    let mut em = event_manager.lock().await;
+
+    match command {
+        SyncCommand::Initialize(rpc_url, contract_hash, completion_ack) => {
+            em.initialize(&rpc_url, &contract_hash).await?;
+            let _ = completion_ack.send(());
+        }
+        SyncCommand::TriggerSync(completion_ack) => {
+            em.process_new_events().await?;
+            let _ = completion_ack.send(());
+        }
+    }
+
+    Ok(())
+}