feat(zksync_cli): Health checkpoint improvements

core/bin/external_node/src/metrics/framework.rs

@@ -5,7 +5,7 @@ use zksync_node_framework::{
     implementations::resources::pools::{MasterPool, PoolResource},
     FromContext, IntoContext, StopReceiver, Task, TaskId, WiringError, WiringLayer,
 };
-use zksync_shared_metrics::rustc::RUST_METRICS;
+use zksync_shared_metrics::binary::BIN_METRICS;
 use zksync_types::{L1ChainId, L2ChainId, SLChainId};
 
 use super::EN_METRICS;
@@ -39,7 +39,7 @@ impl WiringLayer for ExternalNodeMetricsLayer {
     }
 
     async fn wire(self, input: Self::Input) -> Result<Self::Output, WiringError> {
-        RUST_METRICS.initialize();
+        BIN_METRICS.initialize();
         EN_METRICS.observe_config(
             self.l1_chain_id,
             self.sl_chain_id,

core/bin/external_node/src/node_builder.rs

@@ -33,7 +33,7 @@ use zksync_node_framework::{
             NodeStorageInitializerLayer,
         },
         pools_layer::PoolsLayerBuilder,
-        postgres_metrics::PostgresMetricsLayer,
+        postgres::PostgresLayer,
         prometheus_exporter::PrometheusExporterLayer,
         pruning::PruningLayer,
         query_eth_client::QueryEthClientLayer,
@@ -125,8 +125,8 @@ impl ExternalNodeBuilder {
         Ok(self)
     }
 
-    fn add_postgres_metrics_layer(mut self) -> anyhow::Result<Self> {
-        self.node.add_layer(PostgresMetricsLayer);
+    fn add_postgres_layer(mut self) -> anyhow::Result<Self> {
+        self.node.add_layer(PostgresLayer);
         Ok(self)
     }
 
@@ -582,7 +582,7 @@ impl ExternalNodeBuilder {
             // so until we have a dedicated component for "auxiliary" tasks,
             // it's responsible for things like metrics.
             self = self
-                .add_postgres_metrics_layer()?
+                .add_postgres_layer()?
                 .add_external_node_metrics_layer()?;
             // We assign the storage initialization to the core, as it's considered to be
             // the "main" component.

core/bin/zksync_server/src/node_builder.rs

@@ -48,7 +48,7 @@ use zksync_node_framework::{
         object_store::ObjectStoreLayer,
         pk_signing_eth_client::PKSigningEthClientLayer,
         pools_layer::PoolsLayerBuilder,
-        postgres_metrics::PostgresMetricsLayer,
+        postgres::PostgresLayer,
         prometheus_exporter::PrometheusExporterLayer,
         proof_data_handler::ProofDataHandlerLayer,
         query_eth_client::QueryEthClientLayer,
@@ -138,8 +138,8 @@ impl MainNodeBuilder {
         Ok(self)
     }
 
-    fn add_postgres_metrics_layer(mut self) -> anyhow::Result<Self> {
-        self.node.add_layer(PostgresMetricsLayer);
+    fn add_postgres_layer(mut self) -> anyhow::Result<Self> {
+        self.node.add_layer(PostgresLayer);
         Ok(self)
     }
 
@@ -760,9 +760,7 @@ impl MainNodeBuilder {
                     self = self.add_eth_tx_manager_layer()?;
                 }
                 Component::Housekeeper => {
-                    self = self
-                        .add_house_keeper_layer()?
-                        .add_postgres_metrics_layer()?;
+                    self = self.add_house_keeper_layer()?.add_postgres_layer()?;
                 }
                 Component::ProofDataHandler => {
                     self = self.add_proof_data_handler_layer()?;

core/lib/dal/src/eth_sender_dal.rs

@@ -669,7 +669,7 @@ impl EthSenderDal<'_, '_> {
         Ok(())
     }
 
-    pub async fn get_number_of_failed_transactions(&mut self) -> anyhow::Result<i64> {
+    pub async fn get_number_of_failed_transactions(&mut self) -> anyhow::Result<u64> {
         sqlx::query!(
             r#"
             SELECT
@@ -683,6 +683,7 @@ impl EthSenderDal<'_, '_> {
         .fetch_one(self.storage.conn())
         .await?
         .count
+        .map(|c| c as u64)
         .context("count field is missing")
     }
 

core/lib/dal/src/system_dal.rs

@@ -1,5 +1,7 @@
 use std::{collections::HashMap, time::Duration};
 
+use chrono::DateTime;
+use serde::{Deserialize, Serialize};
 use zksync_db_connection::{connection::Connection, error::DalResult, instrument::InstrumentExt};
 
 use crate::Core;
@@ -12,6 +14,16 @@ pub(crate) struct TableSize {
     pub total_size: u64,
 }
 
+#[derive(Debug, Serialize, Deserialize)]
+pub struct DatabaseMigration {
+    pub version: i64,
+    pub description: String,
+    pub installed_on: DateTime<chrono::Utc>,
+    pub success: bool,
+    pub checksum: String,
+    pub execution_time: Duration,
+}
+
 #[derive(Debug)]
 pub struct SystemDal<'a, 'c> {
     pub(crate) storage: &'a mut Connection<'c, Core>,
@@ -86,4 +98,27 @@ impl SystemDal<'_, '_> {
         });
         Ok(table_sizes.collect())
     }
+
+    pub async fn get_last_migration(&mut self) -> DalResult<DatabaseMigration> {
+        let row = sqlx::query!(
+            r#"
+            SELECT *
+            FROM _sqlx_migrations
+            ORDER BY _sqlx_migrations.version DESC
+            LIMIT 1
+            "#
+        )
+        .instrument("get_last_migration")
+        .fetch_one(self.storage)
+        .await?;
+
+        Ok(DatabaseMigration {
+            version: row.version,
+            description: row.description,
+            installed_on: row.installed_on,
+            success: row.success,
+            checksum: hex::encode(row.checksum),
+            execution_time: Duration::from_millis(u64::try_from(row.execution_time).unwrap_or(0)),
+        })
+    }
 }

core/lib/types/src/aggregated_operations.rs

@@ -1,6 +1,8 @@
 use std::{fmt, str::FromStr};
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
 pub enum AggregatedActionType {
     Commit,
     PublishProofOnchain,

core/node/consensus/src/testonly.rs

@@ -584,15 +584,14 @@ impl StateKeeperRunner {
                 let stop_recv = stop_recv.clone();
                 async {
                     ZkSyncStateKeeper::new(
-                        stop_recv,
                         Box::new(io),
                         Box::new(executor_factory),
                         OutputHandler::new(Box::new(persistence.with_tx_insertion()))
                             .with_handler(Box::new(self.sync_state.clone())),
                         Arc::new(NoopSealer),
                         Arc::new(async_cache),
                     )
-                    .run()
+                    .run(stop_recv)
                     .await
                     .context("ZkSyncStateKeeper::run()")?;
                     Ok(())
@@ -665,7 +664,6 @@ impl StateKeeperRunner {
                 let stop_recv = stop_recv.clone();
                 async {
                     ZkSyncStateKeeper::new(
-                        stop_recv,
                         Box::new(io),
                         Box::new(MockBatchExecutor),
                         OutputHandler::new(Box::new(persistence.with_tx_insertion()))
@@ -674,7 +672,7 @@ impl StateKeeperRunner {
                         Arc::new(NoopSealer),
                         Arc::new(MockReadStorageFactory),
                     )
-                    .run()
+                    .run(stop_recv)
                     .await
                     .context("ZkSyncStateKeeper::run()")?;
                     Ok(())

core/node/eth_sender/Cargo.toml

@@ -11,12 +11,14 @@ keywords.workspace = true
 categories.workspace = true
 
 [dependencies]
+serde.workspace = true
 vise.workspace = true
 zksync_types.workspace = true
 zksync_dal.workspace = true
 zksync_config.workspace = true
 zksync_contracts.workspace = true
 zksync_eth_client.workspace = true
+zksync_health_check.workspace = true
 zksync_l1_contract_interface.workspace = true
 zksync_object_store.workspace = true
 zksync_prover_interface.workspace = true

core/node/eth_sender/src/eth_tx_aggregator.rs

@@ -3,6 +3,7 @@ use zksync_config::configs::eth_sender::SenderConfig;
 use zksync_contracts::BaseSystemContractsHashes;
 use zksync_dal::{Connection, ConnectionPool, Core, CoreDal};
 use zksync_eth_client::{BoundEthInterface, CallFunctionArgs};
+use zksync_health_check::{Health, HealthStatus, HealthUpdater, ReactiveHealthCheck};
 use zksync_l1_contract_interface::{
     i_executor::{
         commit::kzg::{KzgInfo, ZK_SYNC_BYTES_PER_BLOB},
@@ -27,6 +28,7 @@ use zksync_types::{
 
 use super::aggregated_operations::AggregatedOperation;
 use crate::{
+    health::{EthTxAggregatorHealthDetails, EthTxDetails},
     metrics::{PubdataKind, METRICS},
     utils::agg_l1_batch_base_cost,
     zksync_functions::ZkSyncFunctions,
@@ -65,6 +67,7 @@ pub struct EthTxAggregator {
     pool: ConnectionPool<Core>,
     settlement_mode: SettlementMode,
     sl_chain_id: SLChainId,
+    health_updater: HealthUpdater,
 }
 
 struct TxData {
@@ -119,10 +122,14 @@ impl EthTxAggregator {
             pool,
             settlement_mode,
             sl_chain_id,
+            health_updater: ReactiveHealthCheck::new("eth_tx_aggregator").1,
         }
     }
 
     pub async fn run(mut self, stop_receiver: watch::Receiver<bool>) -> anyhow::Result<()> {
+        self.health_updater
+            .update(Health::from(HealthStatus::Ready));
+
         let pool = self.pool.clone();
         loop {
             let mut storage = pool.connection_tagged("eth_sender").await.unwrap();
@@ -431,6 +438,13 @@ impl EthTxAggregator {
                 )
                 .await?;
             Self::report_eth_tx_saving(storage, &agg_op, &tx).await;
+
+            self.health_updater.update(
+                EthTxAggregatorHealthDetails {
+                    last_saved_tx: EthTxDetails::new(&tx, None),
+                }
+                .into(),
+            );
         }
         Ok(())
     }
@@ -670,4 +684,9 @@ impl EthTxAggregator {
             )
         })
     }
+
+    /// Returns the health check for eth tx aggregator.
+    pub fn health_check(&self) -> ReactiveHealthCheck {
+        self.health_updater.subscribe()
+    }
 }