feat: Metrics for block tip (#1144)

## What ❔

Add metrics for block tip

## Why ❔
## Checklist

<!-- Check your PR fulfills the following items. -->
<!-- For draft PRs check the boxes as you complete them. -->

- [ ] PR title corresponds to the body of PR (we generate changelog
entries from PRs).
- [ ] Tests for the changes have been added / updated.
- [ ] Documentation comments have been added / updated.
- [ ] Code has been formatted via `zk fmt` and `zk lint`.
- [ ] Spellcheck has been run via `zk spellcheck`.
- [ ] Linkcheck has been run via `zk linkcheck`.

core/lib/zksync_core/src/state_keeper/batch_executor/main_executor.rs

@@ -21,7 +21,7 @@ use super::{BatchExecutor, BatchExecutorHandle, Command, TxExecutionResult};
 use crate::{
     metrics::{InteractionType, TxStage, APP_METRICS},
     state_keeper::{
-        metrics::{TxExecutionStage, EXECUTOR_METRICS, KEEPER_METRICS},
+        metrics::{TxExecutionStage, BATCH_TIP_METRICS, EXECUTOR_METRICS, KEEPER_METRICS},
         types::ExecutionMetricsForCriteria,
     },
 };
@@ -271,6 +271,8 @@ impl CommandReceiver {
                 result.block_tip_execution_result.result
             );
         }
+
+        BATCH_TIP_METRICS.observe(&result.block_tip_execution_result);
         result
     }
 

core/lib/zksync_core/src/state_keeper/keeper.rs

@@ -26,7 +26,10 @@ use super::{
     types::ExecutionMetricsForCriteria,
     updates::UpdatesManager,
 };
-use crate::{gas_tracker::gas_count_from_writes, state_keeper::io::fee_address_migration};
+use crate::{
+    gas_tracker::gas_count_from_writes,
+    state_keeper::{io::fee_address_migration, metrics::BATCH_TIP_METRICS},
+};
 
 /// Amount of time to block on waiting for some resource. The exact value is not really important,
 /// we only need it to not block on waiting indefinitely and be able to process cancellation requests.
@@ -706,6 +709,13 @@ impl ZkSyncStateKeeper {
                 let block_writes_metrics = updates_manager
                     .storage_writes_deduplicator
                     .apply_and_rollback(logs_to_apply_iter.clone());
+
+                BATCH_TIP_METRICS.observe_writes_metrics(
+                    &updates_manager.storage_writes_deduplicator.metrics(),
+                    &block_writes_metrics,
+                    updates_manager.protocol_version(),
+                );
+
                 let block_writes_l1_gas = gas_count_from_writes(
                     &block_writes_metrics,
                     updates_manager.protocol_version(),

core/lib/zksync_core/src/state_keeper/metrics.rs

@@ -5,11 +5,13 @@ use std::{
     time::Duration,
 };
 
+use multivm::interface::VmExecutionResultAndLogs;
 use vise::{
     Buckets, Counter, EncodeLabelSet, EncodeLabelValue, Family, Gauge, Histogram, LatencyObserver,
     Metrics,
 };
 use zksync_mempool::MempoolStore;
+use zksync_types::{tx::tx_execution_info::DeduplicatedWritesMetrics, ProtocolVersionId};
 
 use super::seal_criteria::SealResolution;
 use crate::metrics::InteractionType;
@@ -395,3 +397,54 @@ pub(super) struct ExecutorMetrics {
 
 #[vise::register]
 pub(super) static EXECUTOR_METRICS: vise::Global<ExecutorMetrics> = vise::Global::new();
+
+#[derive(Debug, Metrics)]
+#[metrics(prefix = "batch_tip")]
+pub(crate) struct BatchTipMetrics {
+    #[metrics(buckets = Buckets::exponential(60000.0..=80000000.0, 2.0))]
+    gas_used: Histogram<usize>,
+    #[metrics(buckets = Buckets::exponential(1.0..=60000.0, 2.0))]
+    pubdata_published: Histogram<usize>,
+    #[metrics(buckets = Buckets::exponential(1.0..=4096.0, 2.0))]
+    circuit_statistic: Histogram<usize>,
+    #[metrics(buckets = Buckets::exponential(1.0..=4096.0, 2.0))]
+    execution_metrics_size: Histogram<usize>,
+    #[metrics(buckets = Buckets::exponential(1.0..=60000.0, 2.0))]
+    block_writes_metrics_positive_size: Histogram<usize>,
+    #[metrics(buckets = Buckets::exponential(1.0..=60000.0, 2.0))]
+    block_writes_metrics_negative_size: Histogram<usize>,
+}
+
+impl BatchTipMetrics {
+    pub fn observe(&self, execution_result: &VmExecutionResultAndLogs) {
+        self.gas_used
+            .observe(execution_result.statistics.gas_used as usize);
+        self.pubdata_published
+            .observe(execution_result.statistics.pubdata_published as usize);
+        self.circuit_statistic
+            .observe(execution_result.statistics.circuit_statistic.total());
+        self.execution_metrics_size
+            .observe(execution_result.get_execution_metrics(None).size());
+    }
+
+    pub fn observe_writes_metrics(
+        &self,
+        initial_writes_metrics: &DeduplicatedWritesMetrics,
+        applied_writes_metrics: &DeduplicatedWritesMetrics,
+        protocol_version_id: ProtocolVersionId,
+    ) {
+        let size_diff = applied_writes_metrics.size(protocol_version_id) as i128
+            - initial_writes_metrics.size(protocol_version_id) as i128;
+
+        if size_diff > 0 {
+            self.block_writes_metrics_positive_size
+                .observe(size_diff as usize);
+        } else {
+            self.block_writes_metrics_negative_size
+                .observe(size_diff.unsigned_abs() as usize);
+        }
+    }
+}
+
+#[vise::register]
+pub(crate) static BATCH_TIP_METRICS: vise::Global<BatchTipMetrics> = vise::Global::new();