zebra_consensus/primitives/
sapling.rs1use core::fmt;
4use std::{
5 future::Future,
6 mem,
7 pin::Pin,
8 task::{Context, Poll},
9};
10
11use futures::{future::BoxFuture, FutureExt, TryFutureExt};
12use once_cell::sync::Lazy;
13use rand::thread_rng;
14use tokio::sync::watch;
15use tower::{util::ServiceFn, Service};
16use tower_batch_control::{Batch, BatchControl, RequestWeight};
17use tower_fallback::Fallback;
18
19use sapling_crypto::{bundle::Authorized, BatchValidator, Bundle};
20use zcash_protocol::value::ZatBalance;
21use zebra_chain::transaction::SigHash;
22
23use crate::groth16::SAPLING;
24
25#[derive(Clone)]
26pub struct Item {
27 bundle: Bundle<Authorized, ZatBalance>,
29 sighash: SigHash,
31}
32
33impl Item {
34 pub fn new(bundle: Bundle<Authorized, ZatBalance>, sighash: SigHash) -> Self {
36 Self { bundle, sighash }
37 }
38}
39
40impl RequestWeight for Item {}
41
42#[derive(Default)]
46pub struct Verifier {
47 batch: BatchValidator,
49
50 tx: watch::Sender<Option<bool>>,
55}
56
57impl fmt::Debug for Verifier {
58 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
59 f.debug_struct("Verifier")
60 .field("batch", &"..")
61 .field("tx", &self.tx)
62 .finish()
63 }
64}
65
66impl Drop for Verifier {
67 fn drop(&mut self) {
72 let batch = mem::take(&mut self.batch);
73 let tx = mem::take(&mut self.tx);
74
75 rayon::spawn_fifo(move || {
77 let (spend_vk, output_vk) = SAPLING.verifying_keys();
78
79 let res = batch.validate(&spend_vk, &output_vk, thread_rng());
81 let _ = tx.send(Some(res));
82 });
83 }
84}
85
86impl Service<BatchControl<Item>> for Verifier {
87 type Response = ();
88 type Error = Box<dyn std::error::Error + Send + Sync>;
89 type Future = Pin<Box<dyn Future<Output = Result<(), Self::Error>> + Send>>;
90
91 fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
92 Poll::Ready(Ok(()))
93 }
94
95 fn call(&mut self, req: BatchControl<Item>) -> Self::Future {
96 match req {
97 BatchControl::Item(item) => {
98 let mut rx = self.tx.subscribe();
99
100 let bundle_check = self
101 .batch
102 .check_bundle(item.bundle, item.sighash.into())
103 .then_some(())
104 .ok_or("invalid Sapling bundle");
105
106 async move {
107 bundle_check?;
108
109 rx.changed()
110 .await
111 .map_err(|_| "verifier was dropped without flushing")
112 .and_then(|_| {
113 rx.borrow()
116 .ok_or("threadpool unexpectedly dropped channel sender")?
117 .then(|| {
118 metrics::counter!("proofs.groth16.verified").increment(1);
119 })
120 .ok_or_else(|| {
121 metrics::counter!("proofs.groth16.invalid").increment(1);
122 "batch verification of Sapling shielded data failed"
123 })
124 })
125 .map_err(Self::Error::from)
126 }
127 .boxed()
128 }
129
130 BatchControl::Flush => {
131 let batch = mem::take(&mut self.batch);
132 let tx = mem::take(&mut self.tx);
133
134 async move {
135 tokio::task::spawn_blocking(move || {
136 let (spend_vk, output_vk) = SAPLING.verifying_keys();
137
138 let res = batch.validate(&spend_vk, &output_vk, thread_rng());
139 tx.send(Some(res))
140 })
141 .await?
142 .map_err(Self::Error::from)
143 }
144 .boxed()
145 }
146 }
147 }
148}
149
150pub fn verify_single(
152 item: Item,
153) -> Pin<Box<dyn Future<Output = Result<(), Box<dyn std::error::Error + Send + Sync>>> + Send>> {
154 async move {
155 let mut verifier = Verifier::default();
156
157 let check = verifier
158 .batch
159 .check_bundle(item.bundle, item.sighash.into())
160 .then_some(())
161 .ok_or("invalid Sapling bundle");
162 check?;
163
164 tokio::task::spawn_blocking(move || {
165 let (spend_vk, output_vk) = SAPLING.verifying_keys();
166
167 mem::take(&mut verifier.batch).validate(&spend_vk, &output_vk, thread_rng())
168 })
169 .await
170 .map_err(|_| "Sapling bundle validation thread panicked")?
171 .then_some(())
172 .ok_or("invalid proof or sig in Sapling bundle")
173 }
174 .map_err(Box::from)
175 .boxed()
176}
177
178pub static VERIFIER: Lazy<
180 Fallback<
181 Batch<Verifier, Item>,
182 ServiceFn<
183 fn(Item) -> BoxFuture<'static, Result<(), Box<dyn std::error::Error + Send + Sync>>>,
184 >,
185 >,
186> = Lazy::new(|| {
187 Fallback::new(
188 Batch::new(
189 Verifier::default(),
190 super::MAX_BATCH_SIZE,
191 None,
192 super::MAX_BATCH_LATENCY,
193 ),
194 tower::service_fn(verify_single),
195 )
196});