Add l1 status to citrea_syncStatus (#1192)

* add ledger_db to get_ethereum_rpc

* add ledger_db to ethereum_rpc::Ethereum, get l1 block numbers

* changed CitreaStatus and add err handling

* fix test

* format code, remove unused imports

* add l1 sync test

* rename CitreaStatus, use finalized l1 height

* make requests async

* switch from spawn to join

* fix error names + use ledger db for head l2

* fix lint

---------

Co-authored-by: eyusufatik <[email protected]>

bin/citrea/src/eth.rs

@@ -3,6 +3,7 @@ use std::sync::Arc;
 
 use anyhow::Context as _;
 use ethereum_rpc::{EthRpcConfig, FeeHistoryCacheConfig, GasPriceOracleConfig};
+use sov_db::ledger_db::LedgerDB;
 use sov_modules_api::default_context::DefaultContext;
 use sov_prover_storage_manager::SnapshotManager;
 use sov_rollup_interface::services::da::DaService;
@@ -13,6 +14,7 @@ use tokio::sync::broadcast;
 pub(crate) fn register_ethereum<Da: DaService>(
     da_service: Arc<Da>,
     storage: ProverStorage<sov_state::DefaultStorageSpec, SnapshotManager>,
+    ledger_db: LedgerDB,
     methods: &mut jsonrpsee::RpcModule<()>,
     sequencer_client_url: Option<String>,
     soft_confirmation_rx: Option<broadcast::Receiver<u64>>,
@@ -30,6 +32,7 @@ pub(crate) fn register_ethereum<Da: DaService>(
         da_service,
         eth_rpc_config,
         storage,
+        ledger_db,
         sequencer_client_url,
         soft_confirmation_rx,
     );

bin/citrea/src/rollup/bitcoin.rs

@@ -84,6 +84,7 @@ impl RollupBlueprint for BitcoinRollup {
         crate::eth::register_ethereum::<Self::DaService>(
             da_service.clone(),
             storage.clone(),
+            ledger_db.clone(),
             &mut rpc_methods,
             sequencer_client_url,
             soft_confirmation_rx,

bin/citrea/src/rollup/mock.rs

@@ -75,6 +75,7 @@ impl RollupBlueprint for MockDemoRollup {
         crate::eth::register_ethereum::<Self::DaService>(
             da_service.clone(),
             storage.clone(),
+            ledger_db.clone(),
             &mut rpc_methods,
             sequencer_client_url,
             soft_confirmation_rx,

bin/citrea/tests/e2e/syncing.rs

@@ -3,7 +3,7 @@ use std::str::FromStr;
 use std::time::Duration;
 
 use citrea_stf::genesis_config::GenesisPaths;
-use ethereum_rpc::CitreaStatus;
+use ethereum_rpc::LayerStatus;
 use reth_primitives::{Address, BlockNumberOrTag};
 use sov_mock_da::{MockAddress, MockDaService, MockDaSpec, MockHash};
 use sov_rollup_interface::da::{DaDataLightClient, DaSpec};
@@ -469,10 +469,9 @@ async fn test_full_node_sync_status() {
 
     wait_for_l2_block(&full_node_test_client, 5, Some(Duration::from_secs(60))).await;
 
-    let status = full_node_test_client.citrea_sync_status().await;
-
-    match status {
-        CitreaStatus::Syncing(syncing) => {
+    let l2_status = full_node_test_client.citrea_sync_status().await.l2_status;
+    match l2_status {
+        LayerStatus::Syncing(syncing) => {
             assert!(syncing.synced_block_number > 0 && syncing.synced_block_number < 300);
             assert_eq!(syncing.head_block_number, 300);
         }
@@ -481,10 +480,30 @@ async fn test_full_node_sync_status() {
 
     wait_for_l2_block(&full_node_test_client, 300, Some(Duration::from_secs(60))).await;
 
-    let status = full_node_test_client.citrea_sync_status().await;
+    let l2_status = full_node_test_client.citrea_sync_status().await.l2_status;
+    match l2_status {
+        LayerStatus::Synced(synced_up_to) => assert_eq!(synced_up_to, 300),
+        _ => panic!("Expected synced status"),
+    }
 
-    match status {
-        CitreaStatus::Synced(synced_up_to) => assert_eq!(synced_up_to, 300),
+    // test l1 sync status
+    let da_service = MockDaService::new(MockAddress::default(), &da_db_dir);
+    for _ in 0..19 {
+        da_service.publish_test_block().await.unwrap();
+    }
+    wait_for_prover_l1_height(&full_node_test_client, 1, Some(Duration::from_secs(60))).await;
+    let l1_status = full_node_test_client.citrea_sync_status().await.l1_status;
+    match l1_status {
+        LayerStatus::Syncing(syncing) => {
+            assert!(syncing.synced_block_number > 0 && syncing.synced_block_number < 20);
+            assert_eq!(syncing.head_block_number, 20);
+        }
+        _ => panic!("Expected syncing status"),
+    }
+    wait_for_prover_l1_height(&full_node_test_client, 20, Some(Duration::from_secs(60))).await;
+    let l1_status = full_node_test_client.citrea_sync_status().await.l1_status;
+    match l1_status {
+        LayerStatus::Synced(synced_up_to) => assert_eq!(synced_up_to, 20),
         _ => panic!("Expected synced status"),
     }
 

bin/citrea/tests/test_client/mod.rs

@@ -9,7 +9,7 @@ use alloy::rpc::types::eth::{Block, Transaction, TransactionReceipt, Transaction
 use alloy::signers::local::PrivateKeySigner;
 use alloy::transports::http::{Http, HyperClient};
 use citrea_evm::{Filter, LogResponse};
-use ethereum_rpc::CitreaStatus;
+use ethereum_rpc::SyncStatus;
 use jsonrpsee::core::client::{ClientT, SubscriptionClientT};
 use jsonrpsee::http_client::{HttpClient, HttpClientBuilder};
 use jsonrpsee::rpc_params;
@@ -721,7 +721,7 @@ impl TestClient {
         block_number.saturating_to()
     }
 
-    pub(crate) async fn citrea_sync_status(&self) -> CitreaStatus {
+    pub(crate) async fn citrea_sync_status(&self) -> SyncStatus {
         self.http_client
             .request("citrea_syncStatus", rpc_params![])
             .await

crates/ethereum-rpc/Cargo.toml

@@ -34,6 +34,7 @@ reth-rpc-types = { workspace = true }
 reth-rpc-types-compat = { workspace = true }
 
 # Sovereign-SDK deps
+sov-db = { path = "../../crates/sovereign-sdk/full-node/db/sov-db" }
 sov-modules-api = { path = "../sovereign-sdk/module-system/sov-modules-api", default-features = false }
 sov-rollup-interface = { path = "../sovereign-sdk/rollup-interface", features = ["native"] }
 

crates/ethereum-rpc/src/ethereum.rs

@@ -8,6 +8,7 @@ use reth_rpc_types::trace::geth::GethTrace;
 use rustc_version_runtime::version;
 use schnellru::{ByLength, LruMap};
 use sequencer_client::SequencerClient;
+use sov_db::ledger_db::LedgerDB;
 use sov_modules_api::WorkingSet;
 use sov_rollup_interface::services::da::DaService;
 use sov_rollup_interface::CITREA_VERSION;
@@ -35,19 +36,22 @@ pub struct Ethereum<C: sov_modules_api::Context, Da: DaService> {
     #[cfg(feature = "local")]
     pub(crate) eth_signer: DevSigner,
     pub(crate) storage: C::Storage,
+    pub(crate) ledger_db: LedgerDB,
     pub(crate) sequencer_client: Option<SequencerClient>,
     pub(crate) web3_client_version: String,
     pub(crate) trace_cache: Mutex<LruMap<u64, Vec<GethTrace>, ByLength>>,
     pub(crate) subscription_manager: Option<SubscriptionManager>,
 }
 
 impl<C: sov_modules_api::Context, Da: DaService> Ethereum<C, Da> {
+    #[allow(clippy::too_many_arguments)]
     pub(crate) fn new(
         da_service: Arc<Da>,
         gas_price_oracle_config: GasPriceOracleConfig,
         fee_history_cache_config: FeeHistoryCacheConfig,
         #[cfg(feature = "local")] eth_signer: DevSigner,
         storage: C::Storage,
+        ledger_db: LedgerDB,
         sequencer_client: Option<SequencerClient>,
         soft_confirmation_rx: Option<broadcast::Receiver<u64>>,
     ) -> Self {
@@ -72,6 +76,7 @@ impl<C: sov_modules_api::Context, Da: DaService> Ethereum<C, Da> {
             #[cfg(feature = "local")]
             eth_signer,
             storage,
+            ledger_db,
             sequencer_client,
             web3_client_version: current_version,
             trace_cache,

crates/ethereum-rpc/src/lib.rs

@@ -19,32 +19,40 @@ use reth_rpc_types::trace::geth::{GethDebugTracingOptions, GethTrace};
 use reth_rpc_types::{FeeHistory, Index};
 use sequencer_client::SequencerClient;
 use serde_json::json;
+use sov_db::ledger_db::{LedgerDB, SharedLedgerOps};
+use sov_modules_api::da::BlockHeaderTrait;
 use sov_modules_api::utils::to_jsonrpsee_error_object;
 use sov_modules_api::WorkingSet;
 use sov_rollup_interface::services::da::DaService;
 use subscription::{handle_logs_subscription, handle_new_heads_subscription};
+use tokio::join;
 use tokio::sync::broadcast;
 use trace::{debug_trace_by_block_number, handle_debug_trace_chain};
 use tracing::info;
 
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize, PartialEq)]
 #[serde(rename_all = "camelCase")]
-pub struct SyncStatus {
+pub struct SyncValues {
     pub head_block_number: u64,
     pub synced_block_number: u64,
 }
-
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize, PartialEq)]
-#[serde(rename_all = "camelCase")]
-pub enum CitreaStatus {
+pub enum LayerStatus {
     Synced(u64),
-    Syncing(SyncStatus),
+    Syncing(SyncValues),
+}
+#[derive(Clone, Debug, serde::Serialize, serde::Deserialize, PartialEq)]
+#[serde(rename_all = "camelCase")]
+pub struct SyncStatus {
+    pub l1_status: LayerStatus,
+    pub l2_status: LayerStatus,
 }
 
 pub fn get_ethereum_rpc<C: sov_modules_api::Context, Da: DaService>(
     da_service: Arc<Da>,
     eth_rpc_config: EthRpcConfig,
     storage: C::Storage,
+    ledger_db: LedgerDB,
     sequencer_client_url: Option<String>,
     soft_confirmation_rx: Option<broadcast::Receiver<u64>>,
 ) -> RpcModule<Ethereum<C, Da>> {
@@ -68,6 +76,7 @@ pub fn get_ethereum_rpc<C: sov_modules_api::Context, Da: DaService>(
         #[cfg(feature = "local")]
         eth_signer,
         storage,
+        ledger_db,
         sequencer_client_url.map(SequencerClient::new),
         soft_confirmation_rx,
     ));
@@ -600,47 +609,69 @@ fn register_rpc_methods<C: sov_modules_api::Context, Da: DaService>(
             },
         )?;
 
-        rpc.register_async_method::<Result<CitreaStatus, ErrorObjectOwned>, _, _>(
+        rpc.register_async_method::<Result<SyncStatus, ErrorObjectOwned>, _, _>(
             "citrea_syncStatus",
             |_, ethereum, _| async move {
                 info!("Full Node: citrea_syncStatus");
 
-                // sequencer client should send it
-                let block_number = ethereum
-                    .sequencer_client
-                    .as_ref()
-                    .unwrap()
-                    .block_number()
-                    .await;
-
-                let head_block_number = match block_number {
+                // sequencer client should send latest l2 height
+                // da service should send latest finalized l1 block header
+                let (sequencer_response, da_response) = join!(
+                    ethereum.sequencer_client.as_ref().unwrap().block_number(),
+                    ethereum.da_service.get_last_finalized_block_header()
+                );
+                // handle sequencer response
+                let l2_head_block_number = match sequencer_response {
                     Ok(block_number) => block_number,
                     Err(e) => match e {
                         jsonrpsee::core::client::Error::Call(e_owned) => return Err(e_owned),
                         _ => return Err(to_jsonrpsee_error_object("SEQUENCER_CLIENT_ERROR", e)),
                     },
                 };
 
-                let evm = Evm::<C>::default();
-                let mut working_set = WorkingSet::<C>::new(ethereum.storage.clone());
+                // get l2 synced block number
 
-                let block =
-                    evm.get_block_by_number(Some(BlockNumberOrTag::Latest), None, &mut working_set);
+                let head_soft_confirmation = ethereum.ledger_db.get_head_soft_confirmation();
 
-                let synced_block_number = match block {
-                    Ok(Some(block)) => block.header.number.unwrap(),
+                let l2_synced_block_number = match head_soft_confirmation {
+                    Ok(Some((height, _))) => height.0,
                     Ok(None) => 0u64,
-                    Err(e) => return Err(e),
+                    Err(e) => return Err(to_jsonrpsee_error_object("LEDGER_DB_ERROR", e)),
+                };
+
+                // handle da service response
+                let l1_head_block_number = match da_response {
+                    Ok(header) => header.height(),
+                    Err(e) => return Err(to_jsonrpsee_error_object("DA_SERVICE_ERROR", e)),
                 };
 
-                if synced_block_number < head_block_number {
-                    Ok::<CitreaStatus, ErrorObjectOwned>(CitreaStatus::Syncing(SyncStatus {
-                        synced_block_number,
-                        head_block_number,
-                    }))
+                // get l1 synced block number
+                let l1_synced_block_number = match ethereum.ledger_db.get_last_scanned_l1_height() {
+                    Ok(Some(slot_number)) => slot_number.0,
+                    Ok(None) => 0u64,
+                    Err(e) => return Err(to_jsonrpsee_error_object("LEDGER_DB_ERROR", e)),
+                };
+
+                let l1_status = if l1_synced_block_number < l1_head_block_number {
+                    LayerStatus::Syncing(SyncValues {
+                        synced_block_number: l1_synced_block_number,
+                        head_block_number: l1_head_block_number,
+                    })
                 } else {
-                    Ok::<CitreaStatus, ErrorObjectOwned>(CitreaStatus::Synced(head_block_number))
-                }
+                    LayerStatus::Synced(l1_head_block_number)
+                };
+                let l2_status = if l2_synced_block_number < l2_head_block_number {
+                    LayerStatus::Syncing(SyncValues {
+                        synced_block_number: l2_synced_block_number,
+                        head_block_number: l2_head_block_number,
+                    })
+                } else {
+                    LayerStatus::Synced(l2_head_block_number)
+                };
+                Ok::<SyncStatus, ErrorObjectOwned>(SyncStatus {
+                    l1_status,
+                    l2_status,
+                })
             },
         )?;
     }