feat(state-keeper): More state keeper metrics (#2224)

## What ❔

Add metrics for better observability

## 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`.

core/node/state_keeper/src/keeper.rs

@@ -491,6 +491,8 @@ impl ZkSyncStateKeeper {
         }
 
         while !self.is_canceled() {
+            let full_latency = KEEPER_METRICS.process_l1_batch_loop_iteration.start();
+
             if self
                 .io
                 .should_seal_l1_batch_unconditionally(updates_manager)
@@ -516,7 +518,7 @@ impl ZkSyncStateKeeper {
                     .map_err(|e| e.context("wait_for_new_l2_block_params"))?;
                 tracing::debug!(
                     "Initialized new L2 block #{} (L1 batch #{}) with timestamp {}",
-                    updates_manager.l2_block.number,
+                    updates_manager.l2_block.number + 1,
                     updates_manager.l1_batch.number,
                     display_timestamp(new_l2_block_params.timestamp)
                 );
@@ -596,8 +598,10 @@ impl ZkSyncStateKeeper {
                      transaction {tx_hash}",
                     updates_manager.l1_batch.number
                 );
+                full_latency.observe();
                 return Ok(());
             }
+            full_latency.observe();
         }
         Err(Error::Canceled)
     }
@@ -674,10 +678,12 @@ impl ZkSyncStateKeeper {
         updates_manager: &mut UpdatesManager,
         tx: Transaction,
     ) -> anyhow::Result<(SealResolution, TxExecutionResult)> {
+        let latency = KEEPER_METRICS.execute_tx_outer_time.start();
         let exec_result = batch_executor
             .execute_tx(tx.clone())
             .await
             .with_context(|| format!("failed executing transaction {:?}", tx.hash()))?;
+        latency.observe();
 
         let latency = KEEPER_METRICS.determine_seal_resolution.start();
         // All of `TxExecutionResult::BootloaderOutOfGasForTx`,

core/node/state_keeper/src/metrics.rs

@@ -95,6 +95,12 @@ pub struct StateKeeperMetrics {
     /// The time it takes to determine seal resolution for each tx.
     #[metrics(buckets = Buckets::LATENCIES)]
     pub determine_seal_resolution: Histogram<Duration>,
+    /// The time it takes for state keeper to wait for tx execution result from batch executor.
+    #[metrics(buckets = Buckets::LATENCIES)]
+    pub execute_tx_outer_time: Histogram<Duration>,
+    /// The time it takes for one iteration of the main loop in `process_l1_batch`.
+    #[metrics(buckets = Buckets::LATENCIES)]
+    pub process_l1_batch_loop_iteration: Histogram<Duration>,
 }
 
 fn vm_revert_reason_as_metric_label(reason: &VmRevertReason) -> &'static str {