zecfoundation
/
zebra
mirror of https://github.com/ZcashFoundation/zebra.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Make the CheckpointVerifier handle partial restarts (#736) 

Also put generic bounds on the BlockVerifier struct,
so we get better compilation errors.
This commit is contained in:
teor 2020-07-24 11:47:48 +10:00 committed by GitHub
parent 80597087b3
commit 2acfcf3a90
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 271 additions and 110 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
zebra-consensus/src/block.rs
View File

@ -24,7 +24,14 @@ use tower::{buffer::Buffer, Service};
use zebra_chain::block::{Block, BlockHeaderHash};
struct BlockVerifier<S> {
struct BlockVerifier<S>
where
S: Service<zebra_state::Request, Response = zebra_state::Response, Error = Error>
+ Send
+ Clone
+ 'static,
S::Future: Send + 'static,
{
/// The underlying `ZebraState`, possibly wrapped in other services.
// TODO: contextual verification
#[allow(dead_code)]

24
zebra-consensus/src/chain.rs
View File

@ -160,9 +160,10 @@ where
}
}
/// Return a chain verification service, using `network` and the provided state
/// service. The network is used to create a block verifier and checkpoint
/// verifier.
/// Return a chain verification service, using `network` and `state_service`.
///
/// Gets the initial tip from the state service, and uses it to create a block
/// verifier and checkpoint verifier.
///
/// This function should only be called once for a particular state service. If
/// you need shared block or checkpoint verfiers, create them yourself, and pass
@ -171,7 +172,7 @@ where
// TODO: revise this interface when we generate our own blocks, or validate
// mempool transactions. We might want to share the BlockVerifier, and we
// might not want to add generated blocks to the state.
pub fn init<S>(
pub async fn init<S>(
network: Network,
state_service: S,
) -> impl Service<
@ -189,10 +190,20 @@ where
+ 'static,
S::Future: Send + 'static,
{
tracing::debug!(?network, "initialising ChainVerifier from network");
let initial_tip = zebra_state::initial_tip(state_service.clone())
.await
.expect("State service poll_ready is Ok");
let height = initial_tip.clone().map(|b| b.coinbase_height()).flatten();
let hash = initial_tip.clone().map(|b| b.hash());
tracing::debug!(
?network,
?height,
?hash,
"initialising ChainVerifier with network and initial tip"
);
let block_verifier = crate::block::init(state_service.clone());
let checkpoint_verifier = CheckpointVerifier::new(network);
let checkpoint_verifier = CheckpointVerifier::new(network, initial_tip);
init_from_verifiers(block_verifier, checkpoint_verifier, state_service)
}
@ -214,7 +225,6 @@ where
/// bugs.
pub fn init_from_verifiers<BV, S>(
block_verifier: BV,
// We use an explcit type, so callers can't accidentally swap the verifiers
checkpoint_verifier: CheckpointVerifier,
state_service: S,
) -> impl Service<

4
zebra-consensus/src/chain/tests.rs
View File

@ -63,7 +63,7 @@ fn verifiers_from_checkpoint_list(
let state_service = zebra_state::in_memory::init();
let block_verifier = crate::block::init(state_service.clone());
let checkpoint_verifier =
crate::checkpoint::CheckpointVerifier::from_checkpoint_list(checkpoint_list);
crate::checkpoint::CheckpointVerifier::from_checkpoint_list(checkpoint_list, None);
let chain_verifier =
super::init_from_verifiers(block_verifier, checkpoint_verifier, state_service.clone());
@ -167,7 +167,7 @@ async fn verify_checkpoint() -> Result<(), Report> {
// Test that the chain::init function works. Most of the other tests use
// init_from_verifiers.
let mut chain_verifier = super::init(Mainnet, zebra_state::in_memory::init());
let mut chain_verifier = super::init(Mainnet, zebra_state::in_memory::init()).await;
/// SPANDOC: Make sure the verifier service is ready
let ready_verifier_service = chain_verifier.ready_and().await.map_err(|e| eyre!(e))?;

77
zebra-consensus/src/checkpoint.rs
View File

@ -88,6 +88,9 @@ pub struct CheckpointVerifier {
/// The checkpoint list for this verifier.
checkpoint_list: CheckpointList,
/// The hash of the initial tip, if any.
initial_tip_hash: Option<BlockHeaderHash>,
// Queued Blocks
//
/// A queue of unverified blocks.
@ -111,20 +114,28 @@ pub struct CheckpointVerifier {
/// Contains non-service utility functions for CheckpointVerifiers.
impl CheckpointVerifier {
/// Return a checkpoint verification service for `network`, using the
/// hard-coded checkpoint list. If `initial_tip` is Some(_), the
/// verifier starts at that initial tip, which does not have to be in the
/// hard-coded checkpoint list.
///
/// This function should be called only once for a particular network, rather
/// than constructing multiple verification services for the same network. To
/// Clone a CheckpointVerifier, you might need to wrap it in a
/// clone a CheckpointVerifier, you might need to wrap it in a
/// `tower::Buffer` service.
pub fn new(network: Network) -> Self {
pub fn new(network: Network, initial_tip: Option<Arc<Block>>) -> Self {
let checkpoint_list = CheckpointList::new(network);
let max_height = checkpoint_list.max_height();
tracing::info!(?max_height, ?network, "initialising CheckpointVerifier");
Self::from_checkpoint_list(checkpoint_list)
let initial_height = initial_tip.clone().map(|b| b.coinbase_height()).flatten();
tracing::info!(
?max_height,
?network,
?initial_height,
"initialising CheckpointVerifier"
);
Self::from_checkpoint_list(checkpoint_list, initial_tip)
}
/// Return a checkpoint verification service using `list`.
/// Return a checkpoint verification service using `list` and `initial_tip`.
///
/// Assumes that the provided genesis checkpoint is correct.
///
@ -136,23 +147,48 @@ impl CheckpointVerifier {
#[allow(dead_code)]
pub(crate) fn from_list(
list: impl IntoIterator<Item = (BlockHeight, BlockHeaderHash)>,
initial_tip: Option<Arc<Block>>,
) -> Result<Self, Error> {
Ok(Self::from_checkpoint_list(CheckpointList::from_list(list)?))
Ok(Self::from_checkpoint_list(
CheckpointList::from_list(list)?,
initial_tip,
))
}
/// Return a checkpoint verification service using `checkpoint_list`.
/// Return a checkpoint verification service using `checkpoint_list` and
/// `initial_tip`.
///
/// Callers should prefer `CheckpointVerifier::new`, which uses the
/// hard-coded checkpoint lists. See `CheckpointVerifier::new` and
/// `CheckpointList::from_list` for more details.
pub(crate) fn from_checkpoint_list(checkpoint_list: CheckpointList) -> Self {
pub(crate) fn from_checkpoint_list(
checkpoint_list: CheckpointList,
initial_tip: Option<Arc<Block>>,
) -> Self {
// All the initialisers should call this function, so we only have to
// change fields or default values in one place.
let (initial_tip_hash, verifier_progress) = match initial_tip {
Some(initial_tip) => {
let initial_height = initial_tip
.coinbase_height()
.expect("Bad initial tip: must have coinbase height");
if initial_height >= checkpoint_list.max_height() {
(None, Progress::FinalCheckpoint)
} else {
(
Some(initial_tip.hash()),
Progress::InitialTip(initial_height),
)
}
}
// We start by verifying the genesis block, by itself
None => (None, Progress::BeforeGenesis),
};
CheckpointVerifier {
checkpoint_list,
initial_tip_hash,
queued: BTreeMap::new(),
// We start by verifying the genesis block, by itself
verifier_progress: Progress::BeforeGenesis,
verifier_progress,
}
}
@ -162,7 +198,8 @@ impl CheckpointVerifier {
/// Return the current verifier's progress.
///
/// If verification has not started yet, returns `BeforeGenesis`.
/// If verification has not started yet, returns `BeforeGenesis`,
/// or `InitialTip(height)` if there were cached verified blocks.
///
/// If verification is ongoing, returns `PreviousCheckpoint(height)`.
/// `height` increases as checkpoints are verified.
@ -178,7 +215,7 @@ impl CheckpointVerifier {
fn current_start_bound(&self) -> Option<Bound<BlockHeight>> {
match self.previous_checkpoint_height() {
BeforeGenesis => Some(Unbounded),
PreviousCheckpoint(height) => Some(Excluded(height)),
InitialTip(height) | PreviousCheckpoint(height) => Some(Excluded(height)),
FinalCheckpoint => None,
}
}
@ -204,7 +241,7 @@ impl CheckpointVerifier {
return WaitingForBlocks;
}
BeforeGenesis => BlockHeight(0),
PreviousCheckpoint(height) => height,
InitialTip(height) | PreviousCheckpoint(height) => height,
FinalCheckpoint => return FinishedVerifying,
};
@ -256,6 +293,10 @@ impl CheckpointVerifier {
fn previous_checkpoint_hash(&self) -> Progress<BlockHeaderHash> {
match self.previous_checkpoint_height() {
BeforeGenesis => BeforeGenesis,
InitialTip(_) => self
.initial_tip_hash
.map(InitialTip)
.expect("initial tip height must have an initial tip hash"),
PreviousCheckpoint(height) => self
.checkpoint_list
.hash(height)
@ -282,10 +323,12 @@ impl CheckpointVerifier {
// Any height is valid
BeforeGenesis => {}
// Greater heights are valid
PreviousCheckpoint(previous_height) if (height <= previous_height) => {
InitialTip(previous_height) | PreviousCheckpoint(previous_height)
if (height <= previous_height) =>
{
Err("block height has already been verified")?
}
PreviousCheckpoint(_) => {}
InitialTip(_) | PreviousCheckpoint(_) => {}
// We're finished, so no checkpoint height is valid
FinalCheckpoint => Err("verification has finished")?,
};
@ -311,6 +354,8 @@ impl CheckpointVerifier {
self.verifier_progress = FinalCheckpoint;
} else if self.checkpoint_list.contains(verified_height) {
self.verifier_progress = PreviousCheckpoint(verified_height);
// We're done with the initial tip hash now
self.initial_tip_hash = None;
}
}
@ -473,7 +518,7 @@ impl CheckpointVerifier {
// like other blocks, the genesis parent hash is set by the
// consensus parameters.
BeforeGenesis => parameters::GENESIS_PREVIOUS_BLOCK_HASH,
PreviousCheckpoint(hash) => hash,
InitialTip(hash) | PreviousCheckpoint(hash) => hash,
FinalCheckpoint => return,
};
// Return early if we're still waiting for more blocks

174
zebra-consensus/src/checkpoint/tests.rs
View File

@ -45,7 +45,7 @@ async fn single_item_checkpoint_list() -> Result<(), Report> {
.collect();
let mut checkpoint_verifier =
CheckpointVerifier::from_list(genesis_checkpoint_list).map_err(|e| eyre!(e))?;
CheckpointVerifier::from_list(genesis_checkpoint_list, None).map_err(|e| eyre!(e))?;
assert_eq!(
checkpoint_verifier.previous_checkpoint_height(),
@ -125,7 +125,7 @@ async fn multi_item_checkpoint_list() -> Result<(), Report> {
.collect();
let mut checkpoint_verifier =
CheckpointVerifier::from_list(checkpoint_list).map_err(|e| eyre!(e))?;
CheckpointVerifier::from_list(checkpoint_list, None).map_err(|e| eyre!(e))?;
assert_eq!(
checkpoint_verifier.previous_checkpoint_height(),
@ -206,11 +206,16 @@ async fn multi_item_checkpoint_list() -> Result<(), Report> {
#[tokio::test]
async fn continuous_blockchain_test() -> Result<(), Report> {
continuous_blockchain().await
continuous_blockchain(None).await?;
for height in 0..=10 {
continuous_blockchain(Some(BlockHeight(height))).await?;
}
Ok(())
}
/// Test a continuous blockchain, restarting verification at `restart_height`.
#[spandoc::spandoc]
async fn continuous_blockchain() -> Result<(), Report> {
async fn continuous_blockchain(restart_height: Option<BlockHeight>) -> Result<(), Report> {
zebra_test::init();
// A continuous blockchain
@ -238,63 +243,32 @@ async fn continuous_blockchain() -> Result<(), Report> {
for b in &[
&zebra_test::vectors::BLOCK_MAINNET_GENESIS_BYTES[..],
&zebra_test::vectors::BLOCK_MAINNET_5_BYTES[..],
&zebra_test::vectors::BLOCK_MAINNET_10_BYTES[..],
&zebra_test::vectors::BLOCK_MAINNET_9_BYTES[..],
] {
let block = Arc::<Block>::zcash_deserialize(*b)?;
let hash: BlockHeaderHash = block.as_ref().into();
checkpoints.push((block.clone(), block.coinbase_height().unwrap(), hash));
}
// The checkpoint list will contain only block 0, 5 and 10
// The checkpoint list will contain only block 0, 5 and 9
let checkpoint_list: BTreeMap<BlockHeight, BlockHeaderHash> = checkpoints
.iter()
.map(|(_block, height, hash)| (*height, *hash))
.collect();
let mut checkpoint_verifier =
CheckpointVerifier::from_list(checkpoint_list).map_err(|e| eyre!(e))?;
/// SPANDOC: Verify blocks, restarting at {?restart_height}
{
let initial_tip = restart_height
.map(|BlockHeight(height)| &blockchain[height as usize].0)
.cloned();
let mut checkpoint_verifier =
CheckpointVerifier::from_list(checkpoint_list, initial_tip).map_err(|e| eyre!(e))?;
// Setup checks
assert_eq!(
checkpoint_verifier.previous_checkpoint_height(),
BeforeGenesis
);
assert_eq!(
checkpoint_verifier.target_checkpoint_height(),
WaitingForBlocks
);
assert_eq!(
checkpoint_verifier.checkpoint_list.max_height(),
BlockHeight(10)
);
let mut handles = FuturesUnordered::new();
// Now verify each block
for (block, height, _hash) in blockchain {
/// SPANDOC: Make sure the verifier service is ready
let ready_verifier_service = checkpoint_verifier
.ready_and()
.map_err(|e| eyre!(e))
.await?;
/// SPANDOC: Set up the future for block {?height}
let verify_future = timeout(
Duration::from_secs(VERIFY_TIMEOUT_SECONDS),
ready_verifier_service.call(block.clone()),
);
/// SPANDOC: spawn verification future in the background
let handle = tokio::spawn(verify_future.in_current_span());
handles.push(handle);
// Execution checks
if height < checkpoint_verifier.checkpoint_list.max_height() {
assert_eq!(
checkpoint_verifier.target_checkpoint_height(),
WaitingForBlocks
);
} else {
// Setup checks
if restart_height
.map(|h| h >= checkpoint_verifier.checkpoint_list.max_height())
.unwrap_or(false)
{
assert_eq!(
checkpoint_verifier.previous_checkpoint_height(),
FinalCheckpoint
@ -303,26 +277,86 @@ async fn continuous_blockchain() -> Result<(), Report> {
checkpoint_verifier.target_checkpoint_height(),
FinishedVerifying
);
} else {
assert_eq!(
checkpoint_verifier.previous_checkpoint_height(),
restart_height.map(InitialTip).unwrap_or(BeforeGenesis)
);
assert_eq!(
checkpoint_verifier.target_checkpoint_height(),
WaitingForBlocks
);
}
}
assert_eq!(
checkpoint_verifier.checkpoint_list.max_height(),
BlockHeight(9)
);
while let Some(result) = handles.next().await {
result??.map_err(|e| eyre!(e))?;
}
let mut handles = FuturesUnordered::new();
// Final checks
assert_eq!(
checkpoint_verifier.previous_checkpoint_height(),
FinalCheckpoint
);
assert_eq!(
checkpoint_verifier.target_checkpoint_height(),
FinishedVerifying
);
assert_eq!(
checkpoint_verifier.checkpoint_list.max_height(),
BlockHeight(10)
);
// Now verify each block
for (block, height, _hash) in blockchain {
if let Some(restart_height) = restart_height {
if height <= restart_height {
continue;
}
}
if height > checkpoint_verifier.checkpoint_list.max_height() {
break;
}
/// SPANDOC: Make sure the verifier service is ready
let ready_verifier_service = checkpoint_verifier
.ready_and()
.map_err(|e| eyre!(e))
.await?;
/// SPANDOC: Set up the future for block {?height}
let verify_future = timeout(
Duration::from_secs(VERIFY_TIMEOUT_SECONDS),
ready_verifier_service.call(block.clone()),
);
/// SPANDOC: spawn verification future in the background
let handle = tokio::spawn(verify_future.in_current_span());
handles.push(handle);
// Execution checks
if height < checkpoint_verifier.checkpoint_list.max_height() {
assert_eq!(
checkpoint_verifier.target_checkpoint_height(),
WaitingForBlocks
);
} else {
assert_eq!(
checkpoint_verifier.previous_checkpoint_height(),
FinalCheckpoint
);
assert_eq!(
checkpoint_verifier.target_checkpoint_height(),
FinishedVerifying
);
}
}
while let Some(result) = handles.next().await {
result??.map_err(|e| eyre!(e))?;
}
// Final checks
assert_eq!(
checkpoint_verifier.previous_checkpoint_height(),
FinalCheckpoint
);
assert_eq!(
checkpoint_verifier.target_checkpoint_height(),
FinishedVerifying
);
assert_eq!(
checkpoint_verifier.checkpoint_list.max_height(),
BlockHeight(9)
);
}
Ok(())
}
@ -349,7 +383,7 @@ async fn block_higher_than_max_checkpoint_fail() -> Result<(), Report> {
.collect();
let mut checkpoint_verifier =
CheckpointVerifier::from_list(genesis_checkpoint_list).map_err(|e| eyre!(e))?;
CheckpointVerifier::from_list(genesis_checkpoint_list, None).map_err(|e| eyre!(e))?;
assert_eq!(
checkpoint_verifier.previous_checkpoint_height(),
@ -424,7 +458,7 @@ async fn wrong_checkpoint_hash_fail() -> Result<(), Report> {
.collect();
let mut checkpoint_verifier =
CheckpointVerifier::from_list(genesis_checkpoint_list).map_err(|e| eyre!(e))?;
CheckpointVerifier::from_list(genesis_checkpoint_list, None).map_err(|e| eyre!(e))?;
assert_eq!(
checkpoint_verifier.previous_checkpoint_height(),
@ -604,7 +638,7 @@ async fn checkpoint_drop_cancel() -> Result<(), Report> {
.collect();
let mut checkpoint_verifier =
CheckpointVerifier::from_list(checkpoint_list).map_err(|e| eyre!(e))?;
CheckpointVerifier::from_list(checkpoint_list, None).map_err(|e| eyre!(e))?;
assert_eq!(
checkpoint_verifier.previous_checkpoint_height(),
@ -688,7 +722,7 @@ async fn hard_coded_mainnet() -> Result<(), Report> {
let hash0: BlockHeaderHash = block0.as_ref().into();
// Use the hard-coded checkpoint list
let mut checkpoint_verifier = CheckpointVerifier::new(Network::Mainnet);
let mut checkpoint_verifier = CheckpointVerifier::new(Network::Mainnet, None);
assert_eq!(
checkpoint_verifier.previous_checkpoint_height(),

15
zebra-consensus/src/checkpoint/types.rs
View File

@ -12,8 +12,18 @@ use Target::*;
pub enum Progress<HeightOrHash> {
/// We have not verified any blocks yet.
BeforeGenesis,
/// We have verified up to and including this initial tip.
///
/// Initial tips might not be one of our hard-coded checkpoints, because we
/// might have:
/// - changed the checkpoint spacing, or
/// - added new checkpoints above the initial tip.
InitialTip(HeightOrHash),
/// We have verified up to and including this checkpoint.
PreviousCheckpoint(HeightOrHash),
/// We have finished verifying.
///
/// The final checkpoint is not included in this variant. The verifier has
@ -33,7 +43,10 @@ impl Ord for Progress<BlockHeight> {
(_, BeforeGenesis) => Ordering::Greater,
(FinalCheckpoint, _) => Ordering::Greater,
(_, FinalCheckpoint) => Ordering::Less,
(PreviousCheckpoint(self_height), PreviousCheckpoint(other_height)) => {
(InitialTip(self_height), InitialTip(other_height))
| (InitialTip(self_height), PreviousCheckpoint(other_height))
| (PreviousCheckpoint(self_height), InitialTip(other_height))
| (PreviousCheckpoint(self_height), PreviousCheckpoint(other_height)) => {
self_height.cmp(other_height)
}
}

21
zebra-state/Cargo.toml
View File

@ -9,20 +9,23 @@ edition = "2018"
[dependencies]
zebra-chain = { path = "../zebra-chain" }
tower = "0.3.1"
futures = "0.3.5"
lazy_static = "1.4.0"
hex = "0.4.2"
sled = "0.34.0"
serde = { version = "1", features = ["serde_derive"] }
color-eyre = "0.5"
dirs = "3.0.1"
hex = "0.4.2"
lazy_static = "1.4.0"
serde = { version = "1", features = ["serde_derive"] }
sled = "0.34.0"
futures = "0.3.5"
tower = "0.3.1"
tracing = "0.1"
tracing-futures = "0.2"
[dev-dependencies]
tokio = { version = "0.2.22", features = ["full"] }
zebra-test = { path = "../zebra-test/" }
once_cell = "1.4"
spandoc = "0.2"
tempdir = "0.3.7"
color-eyre = "0.5"
once_cell = "1.4"
tokio = { version = "0.2.22", features = ["full"] }

50
zebra-state/src/lib.rs
View File

@ -14,9 +14,12 @@
#![doc(html_root_url = "https://doc.zebra.zfnd.org/zebra_state")]
#![warn(missing_docs)]
#![allow(clippy::try_err)]
use color_eyre::eyre::{eyre, Report};
use serde::{Deserialize, Serialize};
use std::path::PathBuf;
use std::{iter, sync::Arc};
use std::{error, iter, sync::Arc};
use tower::{Service, ServiceExt};
use zebra_chain::{
block::{Block, BlockHeaderHash},
types::BlockHeight,
@ -133,6 +136,51 @@ fn block_locator_heights(tip_height: BlockHeight) -> impl Iterator<Item = BlockH
.chain(iter::once(BlockHeight(0)))
}
/// The error type for the State Service.
// TODO(jlusby): Error = Report ?
type Error = Box<dyn error::Error + Send + Sync + 'static>;
/// Get the tip block, using `state`.
///
/// If there is no tip, returns `Ok(None)`.
/// Returns an error if `state.poll_ready` errors.
pub async fn initial_tip<S>(state: S) -> Result<Option<Arc<Block>>, Report>
where
S: Service<Request, Response = Response, Error = Error> + Send + Clone + 'static,
S::Future: Send + 'static,
{
let initial_tip_hash = state
.clone()
.ready_and()
.await
.map_err(|e| eyre!(e))?
.call(Request::GetTip)
.await
.map(|response| match response {
Response::Tip { hash } => hash,
_ => unreachable!("GetTip request can only result in Response::Tip"),
})
.ok();
let initial_tip_block = match initial_tip_hash {
Some(hash) => state
.clone()
.ready_and()
.await
.map_err(|e| eyre!(e))?
.call(Request::GetBlock { hash })
.await
.map(|response| match response {
Response::Block { block } => block,
_ => unreachable!("GetBlock request can only result in Response::Block"),
})
.ok(),
None => None,
};
Ok(initial_tip_block)
}
#[cfg(test)]
mod tests {
use super::*;

7
zebrad/src/commands/start.rs
View File

@ -40,6 +40,10 @@ impl StartCmd {
async fn start(&self) -> Result<(), Report> {
info!(?self, "starting to connect to the network");
let config = app_config();
let state = zebra_state::on_disk::init(config.state.clone());
let verifier = zebra_consensus::chain::init(config.network.network, state.clone()).await;
// The service that our node uses to respond to requests by peers
let node = Buffer::new(
service_fn(|req| async move {
@ -48,10 +52,7 @@ impl StartCmd {
}),
1,
);
let config = app_config();
let state = zebra_state::on_disk::init(config.state.clone());
let (peer_set, _address_book) = zebra_network::init(config.network.clone(), node).await;
let verifier = zebra_consensus::chain::init(config.network.network, state.clone());
let mut syncer = sync::Syncer::new(config.network.network, peer_set, state, verifier);