consensus: add a static redjubjub::VERIFIER with fallback

Cargo.lock

@@ -2872,6 +2872,8 @@ dependencies = [
  "tokio",
  "tower",
  "tower-batch",
+ "tower-fallback",
+ "tower-util",
  "tracing",
  "tracing-error",
  "tracing-futures",

zebra-consensus/Cargo.toml

@@ -8,7 +8,6 @@ edition = "2018"
 [dependencies]
 chrono = "0.4.13"
 color-eyre = "0.5"
-once_cell = "1.4"
 rand = "0.7"
 redjubjub = "0.2"
 
@@ -17,9 +16,12 @@ futures = "0.3.5"
 futures-util = "0.3.5"
 tokio = { version = "0.2.22", features = ["time", "sync", "stream", "tracing"] }
 tower = "0.3"
+tower-util = "0.3"
 tracing = "0.1.18"
 tracing-futures = "0.2.4"
+once_cell = "1.4"
 
+tower-fallback = { path = "../tower-fallback/" }
 tower-batch = { path = "../tower-batch/" }
 zebra-chain = { path = "../zebra-chain" }
 zebra-state = { path = "../zebra-state" }

zebra-consensus/src/primitives.rs

@@ -1,3 +1,8 @@
 //! Asynchronous verification of cryptographic primitives.
 
-pub mod redjubjub;
+pub mod redjubjub;
+
+/// The maximum batch size for any of the batch verifiers.
+const MAX_BATCH_SIZE: usize = 64;
+/// The maximum latency bound for any of the batch verifiers.
+const MAX_BATCH_LATENCY: std::time::Duration = std::time::Duration::from_millis(100);

zebra-consensus/src/primitives/redjubjub.rs

@@ -10,11 +10,44 @@ use std::{
     task::{Context, Poll},
 };
 
+use futures::future::{ready, Ready};
+use once_cell::sync::Lazy;
 use rand::thread_rng;
 use redjubjub::{batch, *};
 use tokio::sync::broadcast::{channel, RecvError, Sender};
 use tower::Service;
-use tower_batch::BatchControl;
+use tower_batch::{Batch, BatchControl};
+use tower_fallback::Fallback;
+use tower_util::ServiceFn;
+
+/// Global batch verification context for RedJubjub signatures.
+///
+/// This service transparently batches contemporaneous signature verifications,
+/// handling batch failures by falling back to individual verification.
+///
+/// Note that making a `Service` call requires mutable access to the service, so
+/// you should call `.clone()` on the global handle to create a local, mutable
+/// handle.
+pub static VERIFIER: Lazy<
+    Fallback<Batch<Verifier, Item>, ServiceFn<fn(Item) -> Ready<Result<(), Error>>>>,
+> = Lazy::new(|| {
+    Fallback::new(
+        Batch::new(
+            Verifier::default(),
+            super::MAX_BATCH_SIZE,
+            super::MAX_BATCH_LATENCY,
+        ),
+        // We want to fallback to individual verification if batch verification
+        // fails, so we need a Service to use. The obvious way to do this would
+        // be to write a closure that returns an async block. But because we
+        // have to specify the type of a static, we need to be able to write the
+        // type of the closure and its return value, and both closures and async
+        // blocks have eldritch types whose names cannot be written. So instead,
+        // we use a Ready to avoid an async block and cast the closure to a
+        // function (which is possible because it doesn't capture any state).
+        tower::service_fn((|item: Item| ready(item.verify_single())) as fn(_) -> _),
+    )
+});
 
 /// RedJubjub signature verifier service
 pub struct Verifier {
@@ -26,10 +59,8 @@ pub struct Verifier {
     tx: Sender<Result<(), Error>>,
 }
 
-#[allow(clippy::new_without_default)]
-impl Verifier {
-    /// Create a new RedJubjubVerifier instance
-    pub fn new() -> Self {
+impl Default for Verifier {
+    fn default() -> Self {
         let batch = batch::Verifier::default();
         // XXX(hdevalence) what's a reasonable choice here?
         let (tx, _) = channel(10);

zebra-consensus/src/primitives/redjubjub/tests.rs

@@ -56,7 +56,7 @@ async fn batch_flushes_on_max_items() -> Result<()> {
 
     // Use a very long max_latency and a short timeout to check that
     // flushing is happening based on hitting max_items.
-    let verifier = Batch::new(Verifier::new(), 10, Duration::from_secs(1000));
+    let verifier = Batch::new(Verifier::default(), 10, Duration::from_secs(1000));
     timeout(Duration::from_secs(5), sign_and_verify(verifier, 100))
         .await?
         .map_err(|e| eyre!(e))?;
@@ -75,7 +75,7 @@ async fn batch_flushes_on_max_latency() -> Result<()> {
 
     // Use a very high max_items and a short timeout to check that
     // flushing is happening based on hitting max_latency.
-    let verifier = Batch::new(Verifier::new(), 100, Duration::from_millis(500));
+    let verifier = Batch::new(Verifier::default(), 100, Duration::from_millis(500));
     timeout(Duration::from_secs(5), sign_and_verify(verifier, 10))
         .await?
         .map_err(|e| eyre!(e))?;