Land verified performance evidence stack

crates/chainstate/src/lib.rs

@@ -183,6 +183,14 @@ impl ApplyAdmission {
 struct TransitionGuard<'a> {
     _transition: TransitionAuthorityGuard<'a>,
     _admission: RwLockReadGuard<'a, ()>,
+    started: quanta::Instant,
+}
+
+impl Drop for TransitionGuard<'_> {
+    fn drop(&mut self) {
+        let transition_held_us = self.started.elapsed().as_micros();
+        tracing::debug!(transition_held_us, "chainstate: transition held");
+    }
 }
 
 fn begin_chain_transition<'a>(
@@ -191,10 +199,16 @@ fn begin_chain_transition<'a>(
 ) -> core::result::Result<TransitionGuard<'a>, ApplyError> {
     let admission_guard = admission.enter()?;
     let transition = chain_transition.lock();
-    admission.ensure_open()?;
+    let started = quanta::Instant::now();
+    if let Err(err) = admission.ensure_open() {
+        let transition_held_us = started.elapsed().as_micros();
+        tracing::debug!(transition_held_us, "chainstate: transition held");
+        return Err(err);
+    }
     Ok(TransitionGuard {
         _transition: transition,
         _admission: admission_guard,
+        started,
     })
 }
 

crates/chainstate/src/window.rs

@@ -337,24 +337,34 @@ fn flush_group(
     committed: &mut Vec<ConnectOutcome>,
     blocks: &[&Block],
 ) -> core::result::Result<(), WindowApplyError> {
+    if group.pending.is_empty() {
+        return Ok(());
+    }
     // The staged prefix is a contiguous suffix starting at `committed.len()`,
     // so the newest staged block is borrowed straight from the caller's apply
     // batch — no clone in chainstate. Empty group: nothing staged, no hook.
     // Failure-prefix flush: the failing block was never staged, so the same
     // index still names the newest successfully staged block.
-    let staged_block = if group.pending.is_empty() {
-        None
-    } else {
-        blocks
-            .get(committed.len() + group.pending.len() - 1)
-            .copied()
-    };
+    let staged_block = blocks
+        .get(committed.len() + group.pending.len() - 1)
+        .copied();
+    let flush_started = quanta::Instant::now();
+    let pending_blocks = group.pending.len();
     match group.flush(handles, staged_block) {
         Ok(flushed) => {
+            let flush_wait_us = flush_started.elapsed().as_micros();
+            tracing::debug!(blocks = flushed.len(), flush_wait_us, "window: group flush");
             committed.extend(flushed);
             Ok(())
         }
         Err(flush_error) => {
+            let flush_wait_us = flush_started.elapsed().as_micros();
+            tracing::debug!(
+                blocks = pending_blocks,
+                flush_wait_us,
+                %flush_error,
+                "window: group flush failed"
+            );
             group.abandon();
             Err(WindowApplyError {
                 applied: committed.len(),

crates/chainstate/tests/unit/apply/observability_tests.rs

@@ -192,6 +192,54 @@ fn retired_group_and_check_timings_stay_off_caller_thread_metrics()
     Ok(())
 }
 
+/// Step-24 execution smoke: a real window commit emits the transition-hold
+/// and group-flush diagnostics on the caller thread.
+///
+/// Drives the production `begin_transition` + `connect_window` path on real
+/// regtest fixtures under the `ProfileFields` dispatch and asserts both
+/// `transition_held_us` and `flush_wait_us` arrive. The transition guard is
+/// dropped inside the capture closure so its `Drop` event reaches the same
+/// dispatch. This measures nothing, writes no ledger, and never touches
+/// `/proc/self/clear_refs`: the `VmHWM` performance gate is neither used nor
+/// bypassed, and no timing ratio or promotion is claimed.
+#[test]
+fn window_commit_emits_transition_and_flush_diagnostics() -> Result<(), Box<dyn std::error::Error>>
+{
+    let genesis = Network::Regtest.genesis_block();
+    let fields = ProfileFields::default();
+    capture_profile(&fields, || -> Result<(), Box<dyn std::error::Error>> {
+        let handles = handles(Network::Regtest, Arc::new(UtxoSet::new()));
+        let first = mined_child(genesis.block_hash(), 1)?;
+        let second = mined_child(first.block_hash(), 2)?;
+        seed_genesis(&handles)?;
+        {
+            let mut tree = handles.block_tree.write();
+            bitcoin_rs_chain::accept_headers(
+                &mut tree,
+                &[first.header, second.header],
+                Network::Regtest,
+                bitcoin_rs_chain::current_unix_seconds(),
+                bitcoin_rs_chain::HeaderValidationMode::LiveAdmission,
+            )?;
+        }
+        let blocks = [&first, &second];
+        let serialized: Vec<bytes::Bytes> = blocks
+            .iter()
+            .map(|block| bytes::Bytes::from(consensus_bytes(*block)))
+            .collect();
+        let transition = handles.begin_transition()?;
+        transition.connect_window(&blocks, &serialized)?;
+        Ok(())
+    })?;
+    for field in ["transition_held_us", "flush_wait_us"] {
+        assert!(
+            fields.0.lock().contains(field),
+            "missing window diagnostic: {field}"
+        );
+    }
+    Ok(())
+}
+
 /// Records fields delivered by real tracing events on the apply caller thread.
 #[derive(Clone, Default)]
 struct ProfileFields(Arc<Mutex<HashSet<String>>>);

crates/mempool/src/submit_package.rs

@@ -10,6 +10,18 @@ use crate::{
 use alloc::vec::Vec;
 use bitcoin_rs_primitives::{Tx, unix_time_secs};
 
+/// Orders an arbitrary-order child-with-parents package for submission.
+///
+/// The returned vec is topologically ordered (independent parents sorted by
+/// txid, child last) but deliberately NOT fully structure-checked: package
+/// weight, intra-package conflicts, and package order are validated by
+/// [`crate::package::check_structure`], which
+/// [`MempoolGateway::submit_package_with_times`] runs as the single
+/// authoritative structural gate (snapshot restore enters there directly, so
+/// the check cannot live here without charging local submission a duplicate
+/// pass). This function stays private with [`MempoolGateway::submit_local_package`]
+/// as its only caller, so its output always flows through that gate and can
+/// never reach admission unchecked.
 fn ordered(mut txs: Vec<Tx>) -> Result<Vec<Tx>, SubmitError> {
     if txs.is_empty() || txs.len() > crate::standardness::MAX_PACKAGE_COUNT {
         return Err(AcceptanceRejectReason::PackageCount.into());

crates/node/benches/evidence.rs

@@ -231,6 +231,64 @@ impl Ledger {
     }
 }
 
+/// Samples a run must hold before its p99 may be read as a tail estimate
+/// (`HPA-13`: no trustworthy p99 from too few samples).
+pub const MIN_SAMPLES_FOR_P99: usize = 100;
+
+/// Order statistics of one raw sample set, in the samples' own unit.
+///
+/// The summary never replaces the raw samples; it is printed beside them.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub struct Quantiles {
+    /// Sample count the statistics were computed from.
+    pub n: usize,
+    /// Median, linearly interpolated between adjacent order statistics.
+    pub p50: u64,
+    /// 95th percentile, interpolated the same way.
+    pub p95: u64,
+    /// 99th percentile, interpolated the same way.
+    pub p99: u64,
+    /// Largest sample.
+    pub max: u64,
+}
+
+impl Quantiles {
+    /// Computes p50/p95/p99/max of `samples`; `None` for an empty set, which
+    /// is a missing measurement, never a zero.
+    #[must_use]
+    pub fn of(samples: &[u64]) -> Option<Self> {
+        let mut sorted = samples.to_vec();
+        sorted.sort_unstable();
+        let max = *sorted.last()?;
+        let at = |percent: u64| -> u64 {
+            // Rank scaled by 100 keeps the interpolation exact in integers.
+            let last_rank = u64::try_from(sorted.len() - 1).unwrap_or(u64::MAX);
+            let scaled = percent.saturating_mul(last_rank);
+            let lower = usize::try_from(scaled / 100).unwrap_or(usize::MAX);
+            let remainder = scaled % 100;
+            let low = sorted[lower];
+            let high = sorted.get(lower + 1).copied().unwrap_or(low);
+            let span = u128::from(high - low);
+            let step = span * u128::from(remainder);
+            // Round half up on the interpolated step.
+            low + u64::try_from((step + 50) / 100).unwrap_or(u64::MAX)
+        };
+        Some(Self {
+            n: sorted.len(),
+            p50: at(50),
+            p95: at(95),
+            p99: at(99),
+            max,
+        })
+    }
+
+    /// Whether `p99` rests on enough samples to be read as a tail estimate.
+    #[must_use]
+    pub const fn p99_supported(&self) -> bool {
+        self.n >= MIN_SAMPLES_FOR_P99
+    }
+}
+
 #[cfg(test)]
 #[expect(clippy::expect_used, reason = "benchmark evidence rejection tests")]
 mod tests {
@@ -415,4 +473,37 @@ mod tests {
         );
         assert_eq!(reparsed.cells.len(), ledger.cells.len());
     }
+
+    #[test]
+    fn quantiles_are_exact_order_statistics_with_a_support_floor() {
+        assert_eq!(Quantiles::of(&[]), None, "an empty set is not a zero");
+
+        let single = Quantiles::of(&[7]).expect("one sample");
+        assert_eq!(
+            (single.p50, single.p95, single.p99, single.max),
+            (7, 7, 7, 7)
+        );
+        assert!(!single.p99_supported());
+
+        // Ranks land on whole samples, so the statistics are the samples.
+        let ranks: Vec<u64> = (0..=100).rev().collect();
+        let exact = Quantiles::of(&ranks).expect("101 samples");
+        assert_eq!(
+            (exact.n, exact.p50, exact.p95, exact.p99, exact.max),
+            (101, 50, 95, 99, 100)
+        );
+        assert!(exact.p99_supported());
+        assert!(
+            !Quantiles::of(&ranks[..99])
+                .expect("99 samples")
+                .p99_supported()
+        );
+
+        // Between samples the order statistic is interpolated, halves up.
+        let spread = Quantiles::of(&[40, 10, 30, 20]).expect("four samples");
+        assert_eq!(
+            (spread.p50, spread.p95, spread.p99, spread.max),
+            (25, 39, 40, 40)
+        );
+    }
 }