wip batch verification

Cargo.toml

@@ -19,3 +19,7 @@ panic = "abort"
 
 [profile.release]
 panic = "abort"
+
+[patch.crates-io]
+bellman = { git = "https://github.com/zcashfoundation/librustzcash.git", branch = "batch-bellman" }
+pairing = { git = "https://github.com/zcashfoundation/librustzcash.git", branch = "batch-bellman" }

zebra-consensus/Cargo.toml

@@ -10,6 +10,8 @@ chrono = "0.4.13"
 color-eyre = "0.5"
 rand = "0.7"
 redjubjub = "0.2"
+bellman = "0.6"
+pairing = "0.16"
 
 metrics = "0.12"
 futures = "0.3.5"

zebra-consensus/src/primitives.rs

@@ -1,8 +1,13 @@
 //! Asynchronous verification of cryptographic primitives.
 
+pub mod groth16;
 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);
+
+/// The size of the buffer in the broadcast channels used by batch verifiers.
+const BROADCAST_BUFFER_SIZE: usize = 10;

zebra-consensus/src/primitives/groth16.rs

@@ -0,0 +1,93 @@
+//! Async Groth16 batch verifier service
+
+use std::{
+    future::Future,
+    mem,
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+use futures::future::{ready, Ready};
+use once_cell::sync::Lazy;
+use rand::thread_rng;
+use tokio::sync::broadcast::{channel, RecvError, Sender};
+use tower::Service;
+use tower_batch::{Batch, BatchControl};
+use tower_fallback::Fallback;
+use tower_util::ServiceFn;
+
+use bellman::{
+    groth16::{batch, PreparedVerifyingKey},
+    VerificationError,
+};
+use pairing::bls12_381::Bls12;
+
+/// A Groth16 verification item, used as the request type of the service.
+pub type Item = batch::Item<Bls12>;
+pub type Error = VerificationError;
+
+// XXX we'd like to have a similar Fallback as for Redjubjub,
+// but it's unclear how to do that without capturing a &PreparedVerifyingKey,
+// which prevents us from casting to a function.
+
+pub struct Verifier {
+    batch: batch::Verifier<Bls12>,
+    // Making this 'static makes managing lifetimes much easier.
+    pvk: &'static PreparedVerifyingKey<Bls12>,
+    tx: Sender<Result<(), Error>>,
+}
+
+impl Verifier {
+    fn new(pvk: &'static PreparedVerifyingKey<Bls12>) -> Self {
+        let batch = batch::Verifier::default();
+        let (tx, _) = channel(super::BROADCAST_BUFFER_SIZE);
+        Self { batch, tx, pvk }
+    }
+}
+
+impl Service<BatchControl<Item>> for Verifier {
+    type Response = ();
+    type Error = Error;
+    type Future = Pin<Box<dyn Future<Output = Result<(), Error>> + Send + 'static>>;
+
+    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        Poll::Ready(Ok(()))
+    }
+
+    fn call(&mut self, req: BatchControl<Item>) -> Self::Future {
+        match req {
+            BatchControl::Item(item) => {
+                tracing::trace!("got item");
+                self.batch.queue(item);
+                let mut rx = self.tx.subscribe();
+                Box::pin(async move {
+                    match rx.recv().await {
+                        Ok(result) => result,
+                        Err(RecvError::Lagged(_)) => {
+                            tracing::error!(
+                                "missed channel updates, BROADCAST_BUFFER_SIZE is too low!!"
+                            );
+                            Err(Error::InvalidProof)
+                        }
+                        Err(RecvError::Closed) => panic!("verifier was dropped without flushing"),
+                    }
+                })
+            }
+
+            BatchControl::Flush => {
+                tracing::trace!("got flush command");
+                let batch = mem::take(&mut self.batch);
+                let _ = self.tx.send(batch.verify(thread_rng(), self.pvk));
+                Box::pin(async { Ok(()) })
+            }
+        }
+    }
+}
+
+impl Drop for Verifier {
+    fn drop(&mut self) {
+        // We need to flush the current batch in case there are still any pending futures.
+        let batch = mem::take(&mut self.batch);
+        let _ = self.tx.send(batch.verify(thread_rng(), self.pvk));
+    }
+}

zebra-consensus/src/primitives/redjubjub.rs

@@ -62,8 +62,7 @@ pub struct Verifier {
 impl Default for Verifier {
     fn default() -> Self {
         let batch = batch::Verifier::default();
-        // XXX(hdevalence) what's a reasonable choice here?
-        let (tx, _) = channel(10);
+        let (tx, _) = channel(super::BROADCAST_BUFFER_SIZE);
         Self { tx, batch }
     }
 }
@@ -90,8 +89,8 @@ impl Service<BatchControl<Item>> for Verifier {
                     match rx.recv().await {
                         Ok(result) => result,
                         Err(RecvError::Lagged(_)) => {
-                            tracing::warn!(
-                                "missed channel updates for the correct signature batch!"
+                            tracing::error!(
+                                "missed channel updates, BROADCAST_BUFFER_SIZE is too low!!"
                             );
                             Err(Error::InvalidSignature)
                         }