perf(state): measure mined block admission timing

crates/zakura-consensus/src/block.rs

@@ -585,20 +585,26 @@ where
 {
     let hash = prepared_block.hash;
     let is_mined_commit = admission.is_some();
+    let commit_start = std::time::Instant::now();
+    let ready_start = std::time::Instant::now();
+    let ready_state_service = state_service
+        .ready()
+        .await
+        .map_err(|source| VerifyBlockError::StateService { source, hash })?;
+    if is_mined_commit {
+        metrics::histogram!("state.semantic_commit.ready_wait.duration_seconds")
+            .record(ready_start.elapsed().as_secs_f64());
+    }
+
     let request = match admission {
         Some(admission) => zs::Request::CommitSemanticallyVerifiedBlockWithAdmission {
             block: prepared_block,
             admission,
+            requested_at: std::time::Instant::now(),
         },
         None => zs::Request::CommitSemanticallyVerifiedBlock(prepared_block),
     };
-    let commit_start = std::time::Instant::now();
-    let response = state_service
-        .ready()
-        .await
-        .map_err(|source| VerifyBlockError::StateService { source, hash })?
-        .call(request)
-        .await;
+    let response = ready_state_service.call(request).await;
     if is_mined_commit {
         metrics::histogram!("mining.contextual_commit.duration_seconds")
             .record(commit_start.elapsed().as_secs_f64());

crates/zakura-state/src/request.rs

@@ -8,6 +8,7 @@ use std::{
         atomic::{AtomicBool, AtomicU8, Ordering},
         Arc,
     },
+    time::Instant,
 };
 
 use tokio::sync::Notify;
@@ -1306,6 +1307,8 @@ pub enum Request {
         block: SemanticallyVerifiedBlock,
         /// The admission notification.
         admission: BlockAdmission,
+        /// When consensus submitted this request to the buffered state service.
+        requested_at: Instant,
     },
 
     /// Commit a checkpointed block to the state, skipping most but not all

crates/zakura-state/src/service.rs

@@ -1169,7 +1169,9 @@ impl StateService {
                     }
                     let send_result = non_finalized_block_write_sender.send(queued_child.into());
 
-                    if let Err(SendError(NonFinalizedWriteMessage::Commit(queued))) = send_result {
+                    if let Err(SendError(NonFinalizedWriteMessage::Commit { queued, .. })) =
+                        send_result
+                    {
                         // If Zebra is shutting down, drop blocks and return an error.
                         Self::send_semantically_verified_block_error(
                             queued,
@@ -1675,16 +1677,29 @@ impl Service<Request> for StateService {
                 .boxed()
             }
 
-            Request::CommitSemanticallyVerifiedBlockWithAdmission { block, admission } => {
+            Request::CommitSemanticallyVerifiedBlockWithAdmission {
+                block,
+                admission,
+                requested_at,
+            } => {
                 let timer = CodeTimer::start();
+                metrics::histogram!("state.semantic_commit.dispatch.duration_seconds")
+                    .record(requested_at.elapsed().as_secs_f64());
+
+                let prequeue_checks_start = Instant::now();
                 self.assert_block_can_be_validated(&block);
                 self.pending_utxos.check_against_ordered(&block.new_outputs);
+                metrics::histogram!("state.semantic_commit.prequeue_checks.duration_seconds")
+                    .record(prequeue_checks_start.elapsed().as_secs_f64());
 
+                let queue_send_start = Instant::now();
                 let rsp_rx = tokio::task::block_in_place(move || {
                     span.in_scope(|| {
                         self.queue_and_commit_to_non_finalized_state(block, Some(admission))
                     })
                 });
+                metrics::histogram!("state.semantic_commit.queue_and_commit.duration_seconds")
+                    .record(queue_send_start.elapsed().as_secs_f64());
 
                 timer.finish_desc("CommitSemanticallyVerifiedBlockWithAdmission");
                 let span = Span::current();

crates/zakura-state/src/service/write.rs

@@ -5,7 +5,7 @@ use std::{
     panic::{catch_unwind, resume_unwind, AssertUnwindSafe},
     path::{Path, PathBuf},
     sync::{Arc, OnceLock},
-    time::Duration,
+    time::{Duration, Instant},
 };
 
 use indexmap::IndexMap;
@@ -1632,7 +1632,12 @@ pub enum NonFinalizedWriteMessage {
     },
     /// A newly downloaded and semantically verified block prepared for
     /// contextual validation and insertion into the non-finalized state.
-    Commit(QueuedSemanticallyVerified),
+    Commit {
+        /// The block, response channel, and optional lifecycle reporter.
+        queued: QueuedSemanticallyVerified,
+        /// The instant immediately before the state service attempted the channel send.
+        queued_at: Instant,
+    },
     /// The hash of a block that should be invalidated and removed from
     /// the non-finalized state, if present.
     Invalidate {
@@ -1649,7 +1654,10 @@ pub enum NonFinalizedWriteMessage {
 
 impl From<QueuedSemanticallyVerified> for NonFinalizedWriteMessage {
     fn from(block: QueuedSemanticallyVerified) -> Self {
-        NonFinalizedWriteMessage::Commit(block)
+        NonFinalizedWriteMessage::Commit {
+            queued: block,
+            queued_at: Instant::now(),
+        }
     }
 }
 
@@ -2538,7 +2546,7 @@ impl WriteBlockWorkerTask {
                     let _ = rsp_tx.send(result);
                     None
                 }
-                NonFinalizedWriteMessage::Commit(queued_child) => Some(queued_child),
+                NonFinalizedWriteMessage::Commit { queued, queued_at } => Some((queued, queued_at)),
                 NonFinalizedWriteMessage::Invalidate { hash, rsp_tx } => {
                     tracing::info!(?hash, "invalidating a block in the non-finalized state");
                     let result = if let Some(writer) = header_chain.as_ref() {
@@ -2603,10 +2611,18 @@ impl WriteBlockWorkerTask {
                 }
             };
 
-            let Some((queued_child, rsp_tx, _admission)) = queued_child_and_rsp_tx else {
+            let Some(((queued_child, rsp_tx, admission), queued_at)) = queued_child_and_rsp_tx
+            else {
                 continue;
             };
 
+            let writer_queue_duration = queued_at.elapsed().as_secs_f64();
+            metrics::histogram!("state.block_writer.queue.duration_seconds")
+                .record(writer_queue_duration);
+            if admission.is_some() {
+                metrics::histogram!("state.block_writer.queue.mined.duration_seconds")
+                    .record(writer_queue_duration);
+            }
             let child_hash = queued_child.hash;
             let parent_hash = queued_child.block.header.previous_block_hash;
             let child_height = queued_child.height;

docs/changelog/unreleased/781.md

@@ -0,0 +1,4 @@
+<!-- changelog: none -->
+
+This PR adds internal mined-block admission metrics and has no operator- or
+crate-consumer-visible effect.