Merge #2201

2201: Improve txs count query r=Deniallugo a=Deniallugo

Signed-off-by: deniallugo <[email protected]>

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

core/bin/zksync_api/src/api_server/rest/mod.rs

@@ -18,6 +18,7 @@ use zksync_config::ZkSyncConfig;
 
 mod forced_exit_requests;
 mod helpers;
+pub mod network_status;
 mod v01;
 pub mod v02;
 
@@ -49,7 +50,7 @@ async fn start_server(
                 &api_v01.config.api.common,
                 api_v01.config.api.private.url.clone(),
             );
-            v02::api_scope(tx_sender, &api_v01.config)
+            v02::api_scope(tx_sender, &api_v01.config, api_v01.network_status.clone())
         };
         App::new()
             .wrap(

core/bin/zksync_api/src/api_server/rest/network_status.rs

@@ -0,0 +1,117 @@
+use futures::channel::mpsc;
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::sync::RwLock;
+use tokio::{runtime::Runtime, time};
+use zksync_storage::ConnectionPool;
+use zksync_types::BlockNumber;
+use zksync_utils::panic_notify::ThreadPanicNotify;
+
+#[derive(Default, Debug, Serialize, Deserialize, Clone)]
+pub struct NetworkStatus {
+    pub next_block_at_max: Option<u64>,
+    pub last_committed: BlockNumber,
+    pub last_verified: BlockNumber,
+    pub total_transactions: u32,
+    pub outstanding_txs: u32,
+    pub mempool_size: u32,
+}
+
+#[derive(Debug, Default, Clone)]
+pub struct SharedNetworkStatus(Arc<RwLock<NetworkStatus>>);
+
+impl SharedNetworkStatus {
+    pub async fn read(&self) -> NetworkStatus {
+        (*self.0.as_ref().read().await).clone()
+    }
+
+    pub(crate) async fn update(
+        &mut self,
+        connection_pool: &ConnectionPool,
+        last_tx_seq_no: i64,
+    ) -> Result<i64, anyhow::Error> {
+        let mut storage = connection_pool.access_storage().await?;
+        let mut transaction = storage.start_transaction().await?;
+        let NetworkStatus {
+            total_transactions, ..
+        } = self.read().await;
+
+        let last_verified = transaction
+            .chain()
+            .block_schema()
+            .get_last_verified_confirmed_block()
+            .await
+            .unwrap_or(BlockNumber(0));
+
+        let last_committed = transaction
+            .chain()
+            .block_schema()
+            .get_last_committed_block()
+            .await
+            .unwrap_or(BlockNumber(0));
+
+        let (total_new_transactions, last_seq_no) = transaction
+            .chain()
+            .stats_schema()
+            .count_total_transactions(last_tx_seq_no)
+            .await
+            .unwrap_or((0, 0));
+
+        let mempool_size = transaction
+            .chain()
+            .mempool_schema()
+            .get_mempool_size()
+            .await
+            .unwrap_or(0);
+
+        let outstanding_txs = transaction
+            .chain()
+            .stats_schema()
+            .count_outstanding_proofs(last_verified)
+            .await
+            .unwrap_or(0);
+
+        transaction.commit().await.unwrap_or_default();
+
+        let status = NetworkStatus {
+            next_block_at_max: None,
+            last_committed,
+            last_verified,
+            total_transactions: total_transactions + total_new_transactions,
+            outstanding_txs,
+            mempool_size,
+        };
+
+        // save status to state
+        *self.0.as_ref().write().await = status;
+        Ok(last_seq_no)
+    }
+    pub fn start_updater_detached(
+        mut self,
+        panic_notify: mpsc::Sender<bool>,
+        connection_pool: ConnectionPool,
+    ) {
+        std::thread::Builder::new()
+            .name("rest-state-updater".to_string())
+            .spawn(move || {
+                let _panic_sentinel = ThreadPanicNotify(panic_notify.clone());
+
+                let runtime = Runtime::new().expect("tokio runtime creation");
+
+                let state_update_task = async move {
+                    let mut timer = time::interval(Duration::from_millis(30000));
+                    let mut last_seq_no = 0;
+                    loop {
+                        timer.tick().await;
+                        match self.update(&connection_pool, last_seq_no).await {
+                            Ok(seq_no) => last_seq_no = seq_no,
+                            Err(_) => vlog::error!("Can't update network status"),
+                        }
+                    }
+                };
+                runtime.block_on(state_update_task);
+            })
+            .expect("State update thread");
+    }
+}

core/bin/zksync_api/src/api_server/rest/v01/api_decl.rs

@@ -1,13 +1,11 @@
 //! Declaration of the API structure.
 
-use crate::api_server::rest::{
-    helpers::*,
-    v01::{caches::Caches, network_status::SharedNetworkStatus},
-};
+use crate::api_server::rest::{helpers::*, v01::caches::Caches};
 use actix_web::error::InternalError;
 use actix_web::{web, HttpResponse, Result as ActixResult};
 use futures::channel::mpsc;
 
+use crate::api_server::rest::network_status::SharedNetworkStatus;
 use zksync_config::ZkSyncConfig;
 use zksync_storage::{
     chain::{

core/bin/zksync_api/src/api_server/rest/v01/mod.rs

@@ -1,5 +1,4 @@
 pub mod api_decl;
 pub mod api_impl;
 pub mod caches;
-pub mod network_status;
 pub mod types;

core/bin/zksync_api/src/api_server/rest/v02/mod.rs

@@ -4,6 +4,7 @@ use actix_web::{
     Scope,
 };
 // Workspace uses
+use crate::api_server::rest::network_status::SharedNetworkStatus;
 use zksync_api_types::v02::ApiVersion;
 use zksync_config::ZkSyncConfig;
 use zksync_types::network::Network;
@@ -31,7 +32,11 @@ pub struct SharedData {
     pub api_version: ApiVersion,
 }
 
-pub(crate) fn api_scope(tx_sender: TxSender, zk_config: &ZkSyncConfig) -> Scope {
+pub(crate) fn api_scope(
+    tx_sender: TxSender,
+    zk_config: &ZkSyncConfig,
+    shared_network_status: SharedNetworkStatus,
+) -> Scope {
     let data = SharedData {
         net: zk_config.chain.eth.network,
         api_version: ApiVersion::V02,
@@ -49,7 +54,7 @@ pub(crate) fn api_scope(tx_sender: TxSender, zk_config: &ZkSyncConfig) -> Scope
         ))
         .service(config::api_scope(zk_config))
         .service(fee::api_scope(tx_sender.clone()))
-        .service(status::api_scope(tx_sender.pool.clone()))
+        .service(status::api_scope(shared_network_status))
         .service(token::api_scope(
             zk_config,
             tx_sender.pool.clone(),