1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
//! Top-level semantic block verification for Zebra.
//!
//! Verifies blocks using the [`CheckpointVerifier`] or full [`SemanticBlockVerifier`],
//! depending on the config and block height.
//!
//! # Correctness
//!
//! Block and transaction verification requests should be wrapped in a timeout, because:
//! - checkpoint verification waits for previous blocks, and
//! - full block and transaction verification wait for UTXOs from previous blocks.
//!
//! Otherwise, verification of out-of-order and invalid blocks and transactions can hang
//! indefinitely.

use core::fmt;
use std::{
    future::Future,
    pin::Pin,
    task::{Context, Poll},
};

use futures::{FutureExt, TryFutureExt};
use thiserror::Error;
use tokio::{sync::oneshot, task::JoinHandle};
use tower::{buffer::Buffer, util::BoxService, Service, ServiceExt};
use tracing::{instrument, Instrument, Span};

use zebra_chain::{
    block::{self, Height},
    parameters::Network,
};

use zebra_node_services::mempool;
use zebra_state as zs;

use crate::{
    block::{Request, SemanticBlockVerifier, VerifyBlockError},
    checkpoint::{CheckpointList, CheckpointVerifier, VerifyCheckpointError},
    error::TransactionError,
    transaction, BoxError, Config, ParameterCheckpoint as _,
};

#[cfg(test)]
mod tests;

/// The bound for the chain verifier and transaction verifier buffers.
///
/// We choose the verifier buffer bound based on the maximum number of
/// concurrent verifier users, to avoid contention:
///   - the `ChainSync` block download and verify stream
///   - the `Inbound` block download and verify stream
///   - the `Mempool` transaction download and verify stream
///   - a block miner component, which we might add in future, and
///   - 1 extra slot to avoid contention.
///
/// We deliberately add extra slots, because they only cost a small amount of
/// memory, but missing slots can significantly slow down Zebra.
const VERIFIER_BUFFER_BOUND: usize = 5;

/// The block verifier router routes requests to either the checkpoint verifier or the
/// semantic block verifier, depending on the maximum checkpoint height.
///
/// # Correctness
///
/// Block verification requests should be wrapped in a timeout, so that
/// out-of-order and invalid requests do not hang indefinitely. See the [`router`](`crate::router`)
/// module documentation for details.
struct BlockVerifierRouter<S, V>
where
    S: Service<zs::Request, Response = zs::Response, Error = BoxError> + Send + Clone + 'static,
    S::Future: Send + 'static,
    V: Service<transaction::Request, Response = transaction::Response, Error = BoxError>
        + Send
        + Clone
        + 'static,
    V::Future: Send + 'static,
{
    /// The checkpointing block verifier.
    ///
    /// Always used for blocks before `Canopy`, optionally used for the entire checkpoint list.
    checkpoint: CheckpointVerifier<S>,

    /// The highest permitted checkpoint block.
    ///
    /// This height must be in the `checkpoint` verifier's checkpoint list.
    max_checkpoint_height: block::Height,

    /// The full semantic block verifier, used for blocks after `max_checkpoint_height`.
    block: SemanticBlockVerifier<S, V>,
}

/// An error while semantically verifying a block.
//
// One or both of these error variants are at least 140 bytes
#[derive(Debug, Error)]
#[allow(missing_docs)]
pub enum RouterError {
    /// Block could not be checkpointed
    Checkpoint { source: Box<VerifyCheckpointError> },
    /// Block could not be full-verified
    Block { source: Box<VerifyBlockError> },
}

impl fmt::Display for RouterError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(&match self {
            RouterError::Checkpoint { source } => {
                format!("block could not be checkpointed due to: {source}")
            }
            RouterError::Block { source } => {
                format!("block could not be full-verified due to: {source}")
            }
        })
    }
}

impl From<VerifyCheckpointError> for RouterError {
    fn from(err: VerifyCheckpointError) -> Self {
        RouterError::Checkpoint {
            source: Box::new(err),
        }
    }
}

impl From<VerifyBlockError> for RouterError {
    fn from(err: VerifyBlockError) -> Self {
        RouterError::Block {
            source: Box::new(err),
        }
    }
}

impl RouterError {
    /// Returns `true` if this is definitely a duplicate request.
    /// Some duplicate requests might not be detected, and therefore return `false`.
    pub fn is_duplicate_request(&self) -> bool {
        match self {
            RouterError::Checkpoint { source, .. } => source.is_duplicate_request(),
            RouterError::Block { source, .. } => source.is_duplicate_request(),
        }
    }
}

impl<S, V> Service<Request> for BlockVerifierRouter<S, V>
where
    S: Service<zs::Request, Response = zs::Response, Error = BoxError> + Send + Clone + 'static,
    S::Future: Send + 'static,
    V: Service<transaction::Request, Response = transaction::Response, Error = BoxError>
        + Send
        + Clone
        + 'static,
    V::Future: Send + 'static,
{
    type Response = block::Hash;
    type Error = RouterError;
    type Future =
        Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'static>>;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        // CORRECTNESS
        //
        // The current task must be scheduled for wakeup every time we return
        // `Poll::Pending`.
        //
        // If either verifier is unready, this task is scheduled for wakeup when it becomes
        // ready.
        //
        // We acquire checkpoint readiness before block readiness, to avoid an unlikely
        // hang during the checkpoint to block verifier transition. If the checkpoint and
        // block verifiers are contending for the same buffer/batch, we want the checkpoint
        // verifier to win, so that checkpoint verification completes, and block verification
        // can start. (Buffers and batches have multiple slots, so this contention is unlikely.)
        use futures::ready;
        // The chain verifier holds one slot in each verifier, for each concurrent task.
        // Therefore, any shared buffers or batches polled by these verifiers should double
        // their bounds. (For example, the state service buffer.)
        ready!(self.checkpoint.poll_ready(cx))?;
        ready!(self.block.poll_ready(cx))?;
        Poll::Ready(Ok(()))
    }

    fn call(&mut self, request: Request) -> Self::Future {
        let block = request.block();

        match block.coinbase_height() {
            #[cfg(feature = "getblocktemplate-rpcs")]
            // There's currently no known use case for block proposals below the checkpoint height,
            // so it's okay to immediately return an error here.
            Some(height) if height <= self.max_checkpoint_height && request.is_proposal() => {
                async {
                    // TODO: Add a `ValidateProposalError` enum with a `BelowCheckpoint` variant?
                    Err(VerifyBlockError::ValidateProposal(
                        "block proposals must be above checkpoint height".into(),
                    ))?
                }
                .boxed()
            }

            Some(height) if height <= self.max_checkpoint_height => {
                self.checkpoint.call(block).map_err(Into::into).boxed()
            }
            // This also covers blocks with no height, which the block verifier
            // will reject immediately.
            _ => self.block.call(request).map_err(Into::into).boxed(),
        }
    }
}

/// Initialize block and transaction verification services.
///
/// Returns a block verifier, transaction verifier,
/// a [`BackgroundTaskHandles`] with the state checkpoint verify task,
/// and the maximum configured checkpoint verification height.
///
/// The consensus configuration is specified by `config`, and the Zcash network
/// to verify blocks for is specified by `network`.
///
/// The block verification service asynchronously performs semantic verification
/// checks. Blocks that pass semantic verification are submitted to the supplied
/// `state_service` for contextual verification before being committed to the chain.
///
/// The transaction verification service asynchronously performs semantic verification
/// checks. Transactions that pass semantic verification return an `Ok` result to the caller.
///
/// This function should only be called once for a particular state service.
///
/// Dropped requests are cancelled on a best-effort basis, but may continue to be processed.
///
/// # Correctness
///
/// Block and transaction verification requests should be wrapped in a timeout,
/// so that out-of-order and invalid requests do not hang indefinitely.
/// See the [`router`](`crate::router`) module documentation for details.
#[instrument(skip(state_service, mempool))]
pub async fn init<S, Mempool>(
    config: Config,
    network: &Network,
    mut state_service: S,
    mempool: oneshot::Receiver<Mempool>,
) -> (
    Buffer<BoxService<Request, block::Hash, RouterError>, Request>,
    Buffer<
        BoxService<transaction::Request, transaction::Response, TransactionError>,
        transaction::Request,
    >,
    BackgroundTaskHandles,
    Height,
)
where
    S: Service<zs::Request, Response = zs::Response, Error = BoxError> + Send + Clone + 'static,
    S::Future: Send + 'static,
    Mempool: Service<mempool::Request, Response = mempool::Response, Error = BoxError>
        + Send
        + Clone
        + 'static,
    Mempool::Future: Send + 'static,
{
    // Give other tasks priority before spawning the checkpoint task.
    tokio::task::yield_now().await;

    // Make sure the state contains the known best chain checkpoints, in a separate thread.

    let checkpoint_state_service = state_service.clone();
    let checkpoint_sync = config.checkpoint_sync;
    let checkpoint_network = network.clone();

    let state_checkpoint_verify_handle = tokio::task::spawn(
        // TODO: move this into an async function?
        async move {
            tracing::info!("starting state checkpoint validation");

            // # Consensus
            //
            // We want to verify all available checkpoints, even if the node is not configured
            // to use them for syncing. Zebra's checkpoints are updated with every release,
            // which makes sure they include the latest settled network upgrade.
            //
            // > A network upgrade is settled on a given network when there is a social
            // > consensus that it has activated with a given activation block hash.
            // > A full validator that potentially risks Mainnet funds or displays Mainnet
            // > transaction information to a user MUST do so only for a block chain that
            // > includes the activation block of the most recent settled network upgrade,
            // > with the corresponding activation block hash. Currently, there is social
            // > consensus that NU5 has activated on the Zcash Mainnet and Testnet with the
            // > activation block hashes given in § 3.12 ‘Mainnet and Testnet’ on p. 20.
            //
            // <https://zips.z.cash/protocol/protocol.pdf#blockchain>
            let full_checkpoints = checkpoint_network.checkpoint_list();
            let mut already_warned = false;

            for (height, checkpoint_hash) in full_checkpoints.iter() {
                let checkpoint_state_service = checkpoint_state_service.clone();
                let request = zebra_state::Request::BestChainBlockHash(*height);

                match checkpoint_state_service.oneshot(request).await {
                    Ok(zebra_state::Response::BlockHash(Some(state_hash))) => assert_eq!(
                        *checkpoint_hash, state_hash,
                        "invalid block in state: a previous Zebra instance followed an \
                         incorrect chain. Delete and re-sync your state to use the best chain"
                    ),

                    Ok(zebra_state::Response::BlockHash(None)) => {
                        if checkpoint_sync {
                            tracing::info!(
                                "state is not fully synced yet, remaining checkpoints will be \
                                 verified during syncing"
                            );
                        } else {
                            tracing::warn!(
                                "state is not fully synced yet, remaining checkpoints will be \
                                 verified next time Zebra starts up. Zebra will be less secure \
                                 until it is restarted. Use consensus.checkpoint_sync = true \
                                 in zebrad.toml to make sure you are following a valid chain"
                            );
                        }

                        break;
                    }

                    Ok(response) => {
                        unreachable!("unexpected response type: {response:?} from state request")
                    }
                    Err(e) => {
                        // This error happens a lot in some tests, and it could happen to users.
                        if !already_warned {
                            tracing::warn!(
                                "unexpected error: {e:?} in state request while verifying previous \
                                 state checkpoints. Is Zebra shutting down?"
                            );
                            already_warned = true;
                        }
                    }
                }
            }

            tracing::info!("finished state checkpoint validation");
        }
        .instrument(Span::current()),
    );

    // transaction verification

    let transaction = transaction::Verifier::new(network, state_service.clone(), mempool);
    let transaction = Buffer::new(BoxService::new(transaction), VERIFIER_BUFFER_BOUND);

    // block verification
    let (list, max_checkpoint_height) = init_checkpoint_list(config, network);

    let tip = match state_service
        .ready()
        .await
        .unwrap()
        .call(zs::Request::Tip)
        .await
        .unwrap()
    {
        zs::Response::Tip(tip) => tip,
        _ => unreachable!("wrong response to Request::Tip"),
    };
    tracing::info!(
        ?tip,
        ?max_checkpoint_height,
        "initializing block verifier router"
    );

    let block = SemanticBlockVerifier::new(network, state_service.clone(), transaction.clone());
    let checkpoint = CheckpointVerifier::from_checkpoint_list(list, network, tip, state_service);
    let router = BlockVerifierRouter {
        checkpoint,
        max_checkpoint_height,
        block,
    };

    let router = Buffer::new(BoxService::new(router), VERIFIER_BUFFER_BOUND);

    let task_handles = BackgroundTaskHandles {
        state_checkpoint_verify_handle,
    };

    (router, transaction, task_handles, max_checkpoint_height)
}

/// Parses the checkpoint list for `network` and `config`.
/// Returns the checkpoint list and maximum checkpoint height.
pub fn init_checkpoint_list(config: Config, network: &Network) -> (CheckpointList, Height) {
    // TODO: Zebra parses the checkpoint list three times at startup.
    //       Instead, cache the checkpoint list for each `network`.
    let list = network.checkpoint_list();

    let max_checkpoint_height = if config.checkpoint_sync {
        list.max_height()
    } else {
        list.min_height_in_range(network.mandatory_checkpoint_height()..)
            .expect("hardcoded checkpoint list extends past canopy activation")
    };

    (list, max_checkpoint_height)
}

/// The background task handles for `zebra-consensus` verifier initialization.
#[derive(Debug)]
pub struct BackgroundTaskHandles {
    /// A handle to the state checkpoint verify task.
    /// Finishes when all the checkpoints are verified, or when the state tip is reached.
    pub state_checkpoint_verify_handle: JoinHandle<()>,
}

/// Calls [`init`] with a closed mempool setup channel for conciseness in tests.
///
/// See [`init`] for more details.
#[cfg(any(test, feature = "proptest-impl"))]
pub async fn init_test<S>(
    config: Config,
    network: &Network,
    state_service: S,
) -> (
    Buffer<BoxService<Request, block::Hash, RouterError>, Request>,
    Buffer<
        BoxService<transaction::Request, transaction::Response, TransactionError>,
        transaction::Request,
    >,
    BackgroundTaskHandles,
    Height,
)
where
    S: Service<zs::Request, Response = zs::Response, Error = BoxError> + Send + Clone + 'static,
    S::Future: Send + 'static,
{
    init(
        config.clone(),
        network,
        state_service.clone(),
        oneshot::channel::<
            Buffer<BoxService<mempool::Request, mempool::Response, BoxError>, mempool::Request>,
        >()
        .1,
    )
    .await
}