zebra_consensus/primitives/
sapling.rs

1//! Async Sapling batch verifier service
2
3use core::fmt;
4use std::{
5    future::Future,
6    mem,
7    pin::Pin,
8    task::{Context, Poll},
9};
10
11use futures::{future::BoxFuture, FutureExt};
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_proofs::prover::LocalTxProver;
21use zcash_protocol::value::ZatBalance;
22use zebra_chain::transaction::SigHash;
23
24use crate::{error::TransactionError, BoxError};
25
26/// Sapling prover containing spend and output params for the Sapling circuit.
27///
28/// Used to:
29///
30/// - construct Sapling outputs in coinbase txs, and
31/// - verify Sapling shielded data in the tx verifier.
32static SAPLING: Lazy<LocalTxProver> = Lazy::new(LocalTxProver::bundled);
33
34/// Returns the shared Sapling prover.
35///
36/// Parsing the bundled Sapling parameters takes time, so callers that build Sapling
37/// outputs share this prover instead of parsing the parameters again.
38pub fn prover() -> &'static LocalTxProver {
39    &SAPLING
40}
41
42#[derive(Clone)]
43pub struct Item {
44    /// The bundle containing the Sapling shielded data to verify.
45    bundle: Bundle<Authorized, ZatBalance>,
46    /// The sighash of the transaction that contains the Sapling shielded data.
47    sighash: SigHash,
48}
49
50impl Item {
51    /// Creates a new [`Item`] from a Sapling bundle and sighash.
52    pub fn new(bundle: Bundle<Authorized, ZatBalance>, sighash: SigHash) -> Self {
53        Self { bundle, sighash }
54    }
55}
56
57impl RequestWeight for Item {}
58
59/// A service that verifies Sapling shielded data in batches.
60///
61/// Handles batching incoming requests, driving batches to completion, and reporting results.
62#[derive(Default)]
63pub struct Verifier {
64    /// A batch verifier for Sapling shielded data.
65    batch: BatchValidator,
66
67    /// A channel for broadcasting the verification result of the batch.
68    ///
69    /// Each batch gets a newly created channel, so there is only ever one result sent per channel.
70    /// Tokio doesn't have a oneshot multi-consumer channel, so we use a watch channel.
71    tx: watch::Sender<Option<bool>>,
72}
73
74impl fmt::Debug for Verifier {
75    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
76        f.debug_struct("Verifier")
77            .field("batch", &"..")
78            .field("tx", &self.tx)
79            .finish()
80    }
81}
82
83impl Drop for Verifier {
84    // Flush the current batch in case there are still any pending futures.
85    //
86    // Flushing the batch means we need to validate it. This function fires off the validation and
87    // returns immediately, usually before the validation finishes.
88    fn drop(&mut self) {
89        let batch = mem::take(&mut self.batch);
90        let tx = mem::take(&mut self.tx);
91
92        // The validation is CPU-intensive; do it on a dedicated thread so it does not block.
93        rayon::spawn_fifo(move || {
94            let (spend_vk, output_vk) = SAPLING.verifying_keys();
95
96            // Validate the batch and send the result through the channel.
97            let res = batch.validate(&spend_vk, &output_vk, thread_rng());
98            let _ = tx.send(Some(res));
99        });
100    }
101}
102
103impl Service<BatchControl<Item>> for Verifier {
104    type Response = ();
105    type Error = Box<dyn std::error::Error + Send + Sync>;
106    type Future = Pin<Box<dyn Future<Output = Result<(), Self::Error>> + Send>>;
107
108    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
109        Poll::Ready(Ok(()))
110    }
111
112    fn call(&mut self, req: BatchControl<Item>) -> Self::Future {
113        match req {
114            BatchControl::Item(item) => {
115                let mut rx = self.tx.subscribe();
116
117                let bundle_check = self
118                    .batch
119                    .check_bundle(item.bundle, item.sighash.into())
120                    .then_some(())
121                    .ok_or(TransactionError::SaplingVerificationFailed);
122
123                async move {
124                    bundle_check.map_err(BoxError::from)?;
125
126                    rx.changed()
127                        .await
128                        .map_err(|_| BoxError::from("verifier was dropped without flushing"))?;
129
130                    // We use a new channel for each batch, so we always get the correct
131                    // batch result here.
132                    let is_valid = rx.borrow().ok_or_else(|| {
133                        BoxError::from("threadpool unexpectedly dropped channel sender")
134                    })?;
135
136                    if is_valid {
137                        metrics::counter!("proofs.sapling.verified").increment(1);
138                        Ok(())
139                    } else {
140                        metrics::counter!("proofs.sapling.invalid").increment(1);
141                        Err(BoxError::from(TransactionError::SaplingVerificationFailed))
142                    }
143                }
144                .boxed()
145            }
146
147            BatchControl::Flush => {
148                let batch = mem::take(&mut self.batch);
149                let tx = mem::take(&mut self.tx);
150
151                async move {
152                    let start = std::time::Instant::now();
153                    let spawn_result = tokio::task::spawn_blocking(move || {
154                        let (spend_vk, output_vk) = SAPLING.verifying_keys();
155                        batch.validate(&spend_vk, &output_vk, thread_rng())
156                    })
157                    .await;
158                    let duration = start.elapsed().as_secs_f64();
159
160                    let result_label = match &spawn_result {
161                        Ok(true) => "success",
162                        _ => "failure",
163                    };
164                    metrics::histogram!(
165                        "zebra.consensus.batch.duration_seconds",
166                        "verifier" => "groth16_sapling",
167                        "result" => result_label
168                    )
169                    .record(duration);
170
171                    // Extract the value before consuming spawn_result
172                    let is_valid = spawn_result.as_ref().ok().copied();
173                    let _ = tx.send(is_valid);
174                    spawn_result.map(|_| ()).map_err(Self::Error::from)
175                }
176                .boxed()
177            }
178        }
179    }
180}
181
182/// Verifies a single [`Item`].
183pub fn verify_single(
184    item: Item,
185) -> Pin<Box<dyn Future<Output = Result<(), Box<dyn std::error::Error + Send + Sync>>> + Send>> {
186    async move {
187        let mut verifier = Verifier::default();
188
189        let check = verifier
190            .batch
191            .check_bundle(item.bundle, item.sighash.into())
192            .then_some(())
193            .ok_or(TransactionError::SaplingVerificationFailed);
194        check.map_err(BoxError::from)?;
195
196        let is_valid = tokio::task::spawn_blocking(move || {
197            let (spend_vk, output_vk) = SAPLING.verifying_keys();
198
199            mem::take(&mut verifier.batch).validate(&spend_vk, &output_vk, thread_rng())
200        })
201        .await
202        .map_err(|_| BoxError::from("Sapling bundle validation thread panicked"))?;
203
204        if is_valid {
205            Ok(())
206        } else {
207            Err(BoxError::from(TransactionError::SaplingVerificationFailed))
208        }
209    }
210    .boxed()
211}
212
213/// Global batch verification context for Sapling shielded data.
214pub static VERIFIER: Lazy<
215    Fallback<
216        Batch<Verifier, Item>,
217        ServiceFn<
218            fn(Item) -> BoxFuture<'static, Result<(), Box<dyn std::error::Error + Send + Sync>>>,
219        >,
220    >,
221> = Lazy::new(|| {
222    Fallback::new(
223        Batch::new(
224            Verifier::default(),
225            super::MAX_BATCH_SIZE,
226            None,
227            super::MAX_BATCH_LATENCY,
228        ),
229        tower::service_fn(verify_single),
230    )
231});
zebra_consensus/primitives/sapling.rs

zebra_consensus/primitives/
sapling.rs
