perf(consensus): flush crypto batches at block boundary

CHANGELOG.md

@@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org).
 
 ### Performance
 
+- Explicitly flush consensus crypto batches at the semantic block verification
+  boundary, reducing near-tip block verification latency without changing the
+  global batch latency setting.
 - Compute the v5 ZIP-244 txid and authorizing-data digest natively. Both
   previously routed through `Transaction::to_librustzcash`, which re-serializes
   and reparses the whole transaction — decompressing every Jubjub and Pallas

tower-batch-control/CHANGELOG.md

@@ -5,6 +5,13 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Added
+
+- Added `Batch::flush()` for explicitly queueing a pending batch flush before
+  the configured maximum latency expires.
+
 ## [0.2.42] - 2025-10-15
 
 ### Added

tower-batch-control/src/message.rs

@@ -4,13 +4,19 @@ use tokio::sync::{oneshot, OwnedSemaphorePermit};
 
 use super::error::ServiceError;
 
-/// Message sent to the batch worker
+/// Message sent to the batch worker.
 #[derive(Debug)]
-pub(crate) struct Message<Request, Fut> {
-    pub(crate) request: Request,
-    pub(crate) tx: Tx<Fut>,
-    pub(crate) span: tracing::Span,
-    pub(super) _permit: OwnedSemaphorePermit,
+pub(crate) enum Message<Request, Fut> {
+    /// A batch item request.
+    Item {
+        request: Request,
+        tx: Tx<Fut>,
+        span: tracing::Span,
+        _permit: OwnedSemaphorePermit,
+    },
+
+    /// An explicit request to flush the pending batch.
+    Flush { span: tracing::Span },
 }
 
 /// Response sender

tower-batch-control/src/service.rs

@@ -223,6 +223,19 @@ where
     fn get_worker_error(&self) -> crate::BoxError {
         self.error_handle.get_error_on_closed()
     }
+
+    /// Explicitly flushes the current pending batch.
+    ///
+    /// The flush is queued after item requests that have already been sent to
+    /// this batch worker. This method returns when the command has been queued,
+    /// not when the underlying batch has completed.
+    pub fn flush(&self) -> Result<(), crate::BoxError> {
+        self.tx
+            .send(Message::Flush {
+                span: tracing::Span::current(),
+            })
+            .map_err(|_| self.get_worker_error())
+    }
 }
 
 impl<T, Request: RequestWeight> Service<Request> for Batch<T, Request>
@@ -330,7 +343,7 @@ where
         // acquired, so we can freely allocate a oneshot.
         let (tx, rx) = oneshot::channel();
 
-        match self.tx.send(Message {
+        match self.tx.send(Message::Item {
             request,
             tx,
             span,

tower-batch-control/src/worker.rs

@@ -166,6 +166,12 @@ where
     /// Waits until the inner service is ready,
     /// then stores a future which resolves when the batch finishes.
     async fn flush_service(&mut self) {
+        if self.pending_items_weight == 0 {
+            tracing::trace!("skipping flush for empty batch");
+            self.pending_batch_timer = None;
+            return;
+        }
+
         if self.failed.is_some() {
             tracing::trace!("worker failure: skipping flush");
             return;
@@ -233,18 +239,22 @@ where
                 },
 
                 maybe_msg = self.rx.recv(), if self.can_spawn_new_batches() => match maybe_msg {
-                    Some(msg) => {
+                    Some(Message::Item {
+                        request,
+                        tx,
+                        span,
+                        ..
+                    }) => {
                         tracing::trace!(
                             pending_items_weight = self.pending_items_weight,
                             batch_deadline = ?self.pending_batch_timer.as_ref().map(|sleep| sleep.deadline()),
                             running_batches = self.concurrent_batches.len(),
                             "batch message received",
                         );
 
-                        let span = msg.span;
                         let is_new_batch = self.pending_items_weight == 0;
 
-                        self.process_req(msg.request, msg.tx)
+                        self.process_req(request, tx)
                             // Apply the provided span to request processing.
                             .instrument(span)
                             .await;
@@ -279,6 +289,16 @@ where
                             );
                         }
                     }
+                    Some(Message::Flush { span }) => {
+                        tracing::trace!(
+                            pending_items_weight = self.pending_items_weight,
+                            batch_deadline = ?self.pending_batch_timer.as_ref().map(|sleep| sleep.deadline()),
+                            running_batches = self.concurrent_batches.len(),
+                            "explicit batch flush received",
+                        );
+
+                        self.flush_service().instrument(span).await;
+                    }
                     None => {
                         tracing::trace!("batch channel closed and emptied, exiting worker task");
 
@@ -375,9 +395,9 @@ where
 
         // Fail queued requests
         while let Ok(msg) = self.rx.try_recv() {
-            let _ = msg
-                .tx
-                .send(Err(self.failed.as_ref().expect("just set failed").clone()));
+            if let Message::Item { tx, .. } = msg {
+                let _ = tx.send(Err(self.failed.as_ref().expect("just set failed").clone()));
+            }
         }
 
         // Clear any finished batches, ignoring any errors.

tower-batch-control/tests/ed25519.rs

@@ -223,6 +223,30 @@ where
     Ok(())
 }
 
+async fn sign_and_verify_after_explicit_flush(
+    mut verifier: Batch<Verifier, Item>,
+    n: usize,
+) -> Result<(), BoxError> {
+    let mut results = FuturesOrdered::new();
+    for _ in 0..n {
+        let sk = SigningKey::new(thread_rng());
+        let vk_bytes = VerificationKeyBytes::from(&sk);
+        let msg = b"BatchVerifyTest";
+        let sig = sk.sign(&msg[..]);
+
+        verifier.ready().await?;
+        results.push_back(verifier.call((vk_bytes, sig, msg).into()));
+    }
+
+    verifier.flush()?;
+
+    while let Some(result) = results.next().await {
+        result?;
+    }
+
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread")]
 async fn batch_flushes_on_max_items_weight() -> Result<(), Report> {
     use tokio::time::timeout;
@@ -259,6 +283,47 @@ async fn batch_flushes_on_max_latency() -> Result<(), Report> {
     Ok(())
 }
 
+#[tokio::test(flavor = "multi_thread")]
+async fn batch_flushes_on_explicit_flush() -> Result<(), Report> {
+    use tokio::time::timeout;
+    let _init_guard = zebra_test::init();
+
+    // Use a very high max_items and long max_latency. Without the explicit
+    // flush, this verification would wait for the latency timer.
+    let verifier = Batch::new(Verifier::default(), 100, 10, Duration::from_secs(1000));
+    timeout(
+        Duration::from_secs(1),
+        sign_and_verify_after_explicit_flush(verifier, 10),
+    )
+    .await
+    .map_err(|e| eyre!(e))?
+    .map_err(|e| eyre!(e))?;
+
+    Ok(())
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn batch_flush_on_empty_batch_is_noop() -> Result<(), Report> {
+    use tokio::time::timeout;
+    let _init_guard = zebra_test::init();
+
+    // Flushing with no pending items must not fail the worker or disturb later
+    // batches: items queued after the empty flush are still verified by the
+    // next explicit flush.
+    let verifier = Batch::new(Verifier::default(), 100, 10, Duration::from_secs(1000));
+    verifier.flush().map_err(|e| eyre!(e))?;
+
+    timeout(
+        Duration::from_secs(1),
+        sign_and_verify_after_explicit_flush(verifier, 10),
+    )
+    .await
+    .map_err(|e| eyre!(e))?
+    .map_err(|e| eyre!(e))?;
+
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread")]
 async fn fallback_verification() -> Result<(), Report> {
     let _init_guard = zebra_test::init();

tower-fallback/CHANGELOG.md

@@ -5,6 +5,12 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Added
+
+- Added `Fallback::primary()` for read-only access to the primary service.
+
 ## [0.2.41] - 2025-07-11
 
 First "stable" release. However, be advised that the API may still greatly

tower-fallback/src/service.rs

@@ -31,6 +31,14 @@ impl<S1, S2: Clone> Fallback<S1, S2> {
     pub fn new(svc1: S1, svc2: S2) -> Self {
         Self { svc1, svc2 }
     }
+
+    /// Returns the primary service.
+    ///
+    /// Requests are first attempted using this service, before retrying on the
+    /// fallback service.
+    pub fn primary(&self) -> &S1 {
+        &self.svc1
+    }
 }
 
 impl<S1, S2, Request> Service<Request> for Fallback<S1, S2>

zebra-consensus/src/block.rs

@@ -31,7 +31,7 @@ use zebra_chain::{
 };
 use zebra_state as zs;
 
-use crate::{error::*, transaction as tx, BoxError};
+use crate::{error::*, primitives, transaction as tx, BoxError};
 
 pub mod check;
 pub mod request;
@@ -276,6 +276,14 @@ where
 
             let known_outpoint_hashes: Arc<HashSet<transaction::Hash>> =
                 Arc::new(known_utxos.keys().map(|outpoint| outpoint.hash).collect());
+            // This guard must be declared after `known_outpoint_hashes`: the flush
+            // registry is keyed by the `Arc` allocation address, so the guard must be
+            // dropped first, removing the registry entry while the address is still
+            // pinned by this `Arc` and cannot be reused by another block's key.
+            let _block_batch_flush = primitives::register_block_verifier_batch_flush(
+                &known_outpoint_hashes,
+                block.transactions.len(),
+            );
 
             for (&transaction_hash, transaction) in
                 transaction_hashes.iter().zip(block.transactions.iter())