grin/chain/src/pipe.rs

// Copyright 2018 The Grin Developers
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! Implementation of the chain block acceptance (or refusal) pipeline.

use std::sync::Arc;
use util::RwLock;

use chrono::prelude::Utc;
use chrono::Duration;

use lru_cache::LruCache;

use chain::OrphanBlockPool;
use core::consensus;
use core::core::hash::{Hash, Hashed};
use core::core::verifier_cache::VerifierCache;
use core::core::Committed;
use core::core::{Block, BlockHeader, BlockSums};
use core::global;
use core::pow;
use error::{Error, ErrorKind};
use grin_store;
use store;
use txhashset;
use types::{Options, Tip};

/// Contextual information required to process a new block and either reject or
/// accept it.
pub struct BlockContext<'a> {
	/// The options
	pub opts: Options,
	/// The pow verifier to use when processing a block.
	pub pow_verifier: fn(&BlockHeader, u8) -> Result<(), pow::Error>,
	/// The active txhashset (rewindable MMRs) to use for block processing.
	pub txhashset: &'a mut txhashset::TxHashSet,
	/// The active batch to use for block processing.
	pub batch: store::Batch<'a>,

	/// Recently processed blocks to avoid double-processing
	pub block_hashes_cache: Arc<RwLock<LruCache<Hash, bool>>>,
	/// The verifier cache (caching verifier for rangeproofs and kernel signatures)
	pub verifier_cache: Arc<RwLock<VerifierCache>>,
	/// Recent orphan blocks to avoid double-processing
	pub orphans: Arc<OrphanBlockPool>,
}

/// Process a block header as part of processing a full block.
/// We want to make sure the header is valid before we process the full block.
fn process_header_for_block(header: &BlockHeader, ctx: &mut BlockContext) -> Result<(), Error> {
	let head = ctx.batch.head()?;

	// If we do not have the previous header then treat the block for this header
	// as an orphan.
	if ctx.batch.get_previous_header(header).is_err() {
		return Err(ErrorKind::Orphan.into());
	}

	txhashset::header_extending(&mut ctx.txhashset, &mut ctx.batch, |extension| {
		extension.force_rollback();

		let prev = extension.batch.get_previous_header(header)?;
		if prev.hash() == head.last_block_h {
			// Not a fork so we do not need to rewind or reapply any headers.
		} else {
			// Rewind and re-apply headers on the forked chain to
			// put the header extension in the correct forked state
			// (immediately prior to this new header).
			rewind_and_apply_header_fork(header, extension)?;
		}

		// Check the current root is correct.
		extension.validate_root(header)?;

		// Apply the new header to our header extension.
		extension.apply_header(header)?;

		Ok(())
	})?;

	validate_header(header, ctx)?;
	add_block_header(header, &ctx.batch)?;
	update_header_head(header, ctx)?;

	Ok(())
}

/// Runs the block processing pipeline, including validation and finding a
/// place for the new block in the chain.
/// Returns new head if chain head updated.
pub fn process_block(b: &Block, ctx: &mut BlockContext) -> Result<Option<Tip>, Error> {
	// TODO should just take a promise for a block with a full header so we don't
	// spend resources reading the full block when its header is invalid

	debug!(
		"pipe: process_block {} at {}, in/out/kern: {}/{}/{}",
		b.hash(),
		b.header.height,
		b.inputs().len(),
		b.outputs().len(),
		b.kernels().len(),
	);

	// Fast in-memory checks to avoid re-processing a block we recently processed.
	{
		// Check if we have recently processed this block (via ctx chain head).
		check_known_head(&b.header, ctx)?;

		// Check if we have recently processed this block (via block_hashes_cache).
		check_known_cache(&b.header, ctx)?;

		// Check if this block is already know due it being in the current set of orphan blocks.
		check_known_orphans(&b.header, ctx)?;

		// Check we have *this* block in the store.
		// Stop if we have processed this block previously (it is in the store).
		// This is more expensive than the earlier check_known() as we hit the store.
		check_known_store(&b.header, ctx)?;
	}

	// Header specific processing.
	process_header_for_block(&b.header, ctx)?;

	// Check if are processing the "next" block relative to the current chain head.
	let prev_header = ctx.batch.get_previous_header(&b.header)?;
	let head = ctx.batch.head()?;
	if prev_header.hash() == head.last_block_h {
		// If this is the "next" block then either -
		//   * common case where we process blocks sequentially.
		//   * special case where this is the first fast sync full block
		// Either way we can proceed (and we know the block is new and unprocessed).
	} else {
		// At this point it looks like this is a new block that we have not yet processed.
		// Check we have the *previous* block in the store.
		// If we do not then treat this block as an orphan.
		check_prev_store(&b.header, &mut ctx.batch)?;
	}

	// Validate the block itself, make sure it is internally consistent.
	// Use the verifier_cache for verifying rangeproofs and kernel signatures.
	validate_block(b, ctx)?;

	// Start a chain extension unit of work dependent on the success of the
	// internal validation and saving operations
	txhashset::extending(&mut ctx.txhashset, &mut ctx.batch, |mut extension| {
		let prev = extension.batch.get_previous_header(&b.header)?;
		if prev.hash() == head.last_block_h {
			// Not a fork so we do not need to rewind or reapply any blocks.
		} else {
			// Rewind and re-apply blocks on the forked chain to
			// put the txhashset in the correct forked state
			// (immediately prior to this new block).
			rewind_and_apply_fork(b, extension)?;
		}

		// Check any coinbase being spent have matured sufficiently.
		// This needs to be done within the context of a potentially
		// rewound txhashset extension to reflect chain state prior
		// to applying the new block.
		verify_coinbase_maturity(b, &mut extension)?;

		// Validate the block against the UTXO set.
		validate_utxo(b, &mut extension)?;

		// Using block_sums (utxo_sum, kernel_sum) for the previous block from the db
		// we can verify_kernel_sums across the full UTXO sum and full kernel sum
		// accounting for inputs/outputs/kernels in this new block.
		// We know there are no double-spends etc. if this verifies successfully.
		verify_block_sums(b, &mut extension)?;

		// Apply the block to the txhashset state.
		// Validate the txhashset roots and sizes against the block header.
		// Block is invalid if there are any discrepencies.
		apply_block_to_txhashset(b, &mut extension)?;

		// If applying this block does not increase the work on the chain then
		// we know we have not yet updated the chain to produce a new chain head.
		let head = extension.batch.head()?;
		if !has_more_work(&b.header, &head) {
			extension.force_rollback();
		}

		Ok(())
	})?;

	// Add the validated block to the db.
	// We do this even if we have not increased the total cumulative work
	// so we can maintain multiple (in progress) forks.
	add_block(b, &ctx.batch)?;

	// Update the chain head if total work is increased.
	let res = update_head(b, ctx)?;
	Ok(res)
}

/// Process the block header.
/// This is only ever used during sync and uses a context based on sync_head.
pub fn sync_block_headers(
	headers: &Vec<BlockHeader>,
	ctx: &mut BlockContext,
) -> Result<Option<Tip>, Error> {
	if let Some(header) = headers.first() {
		debug!(
			"pipe: sync_block_headers: {} headers from {} at {}",
			headers.len(),
			header.hash(),
			header.height,
		);
	} else {
		return Ok(None);
	}

	let all_known = if let Some(last_header) = headers.last() {
		ctx.batch.get_block_header(&last_header.hash()).is_ok()
	} else {
		false
	};

	if !all_known {
		let first_header = headers.first().unwrap();
		let prev_header = ctx.batch.get_previous_header(&first_header)?;
		txhashset::sync_extending(&mut ctx.txhashset, &mut ctx.batch, |extension| {
			extension.rewind(&prev_header)?;

			for header in headers {
				// Check the current root is correct.
				extension.validate_root(header)?;

				// Apply the header to the header MMR.
				extension.apply_header(header)?;

				// Save the header to the db.
				add_block_header(header, &extension.batch)?;
			}

			Ok(())
		})?;

		// Validate all our headers now that we have added each "previous"
		// header to the db in this batch above.
		for header in headers {
			validate_header(header, ctx)?;
		}
	}

	// Update header_head (if most work) and sync_head (regardless) in all cases,
	// even if we already know all the headers.
	// This avoids the case of us getting into an infinite loop with sync_head never
	// progressing.
	// We only need to do this once at the end of this batch of headers.
	if let Some(header) = headers.last() {
		// Update sync_head regardless of total work.
		update_sync_head(header, &mut ctx.batch)?;

		// Update header_head (but only if this header increases our total known work).
		// i.e. Only if this header is now the head of the current "most work" chain.
		let res = update_header_head(header, ctx)?;
		Ok(res)
	} else {
		Ok(None)
	}
}

/// Process block header as part of "header first" block propagation.
/// We validate the header but we do not store it or update header head based
/// on this. We will update these once we get the block back after requesting
/// it.
pub fn process_block_header(header: &BlockHeader, ctx: &mut BlockContext) -> Result<(), Error> {
	debug!(
		"pipe: process_block_header: {} at {}",
		header.hash(),
		header.height,
	); // keep this

	check_header_known(header, ctx)?;
	validate_header(header, ctx)?;
	Ok(())
}

/// Quick in-memory check to fast-reject any block header we've already handled
/// recently. Keeps duplicates from the network in check.
/// ctx here is specific to the header_head (tip of the header chain)
fn check_header_known(header: &BlockHeader, ctx: &mut BlockContext) -> Result<(), Error> {
	let header_head = ctx.batch.header_head()?;
	if header.hash() == header_head.last_block_h || header.hash() == header_head.prev_block_h {
		return Err(ErrorKind::Unfit("header already known".to_string()).into());
	}
	Ok(())
}

/// Quick in-memory check to fast-reject any block handled recently.
/// Keeps duplicates from the network in check.
/// Checks against the last_block_h and prev_block_h of the chain head.
fn check_known_head(header: &BlockHeader, ctx: &mut BlockContext) -> Result<(), Error> {
	let head = ctx.batch.head()?;
	let bh = header.hash();
	if bh == head.last_block_h || bh == head.prev_block_h {
		return Err(ErrorKind::Unfit("already known in head".to_string()).into());
	}
	Ok(())
}

/// Quick in-memory check to fast-reject any block handled recently.
/// Keeps duplicates from the network in check.
/// Checks against the cache of recently processed block hashes.
fn check_known_cache(header: &BlockHeader, ctx: &mut BlockContext) -> Result<(), Error> {
	let mut cache = ctx.block_hashes_cache.write();
	if cache.contains_key(&header.hash()) {
		return Err(ErrorKind::Unfit("already known in cache".to_string()).into());
	}
	Ok(())
}

/// Check if this block is in the set of known orphans.
fn check_known_orphans(header: &BlockHeader, ctx: &mut BlockContext) -> Result<(), Error> {
	if ctx.orphans.contains(&header.hash()) {
		Err(ErrorKind::Unfit("already known in orphans".to_string()).into())
	} else {
		Ok(())
	}
}

// Check if this block is in the store already.
fn check_known_store(header: &BlockHeader, ctx: &mut BlockContext) -> Result<(), Error> {
	match ctx.batch.block_exists(&header.hash()) {
		Ok(true) => {
			let head = ctx.batch.head()?;
			if header.height < head.height.saturating_sub(50) {
				// TODO - we flag this as an "abusive peer" but only in the case
				// where we have the full block in our store.
				// So this is not a particularly exhaustive check.
				Err(ErrorKind::OldBlock.into())
			} else {
				Err(ErrorKind::Unfit("already known in store".to_string()).into())
			}
		}
		Ok(false) => {
			// Not yet processed this block, we can proceed.
			Ok(())
		}
		Err(e) => {
			return Err(ErrorKind::StoreErr(e, "pipe get this block".to_owned()).into());
		}
	}
}

// Check we have the *previous* block in the store.
// Note: not just the header but the full block itself.
// We cannot assume we can use the chain head for this
// as we may be dealing with a fork (with less work currently).
fn check_prev_store(header: &BlockHeader, batch: &mut store::Batch) -> Result<(), Error> {
	let prev = batch.get_previous_header(&header)?;
	match batch.block_exists(&prev.hash()) {
		Ok(true) => {
			// We have the previous block in the store, so we can proceed.
			Ok(())
		}
		Ok(false) => {
			// We do not have the previous block in the store.
			// We have not yet processed the previous block so
			// this block is an orphan (for now).
			Err(ErrorKind::Orphan.into())
		}
		Err(e) => Err(ErrorKind::StoreErr(e, "pipe get previous".to_owned()).into()),
	}
}

/// First level of block validation that only needs to act on the block header
/// to make it as cheap as possible. The different validations are also
/// arranged by order of cost to have as little DoS surface as possible.
fn validate_header(header: &BlockHeader, ctx: &mut BlockContext) -> Result<(), Error> {
	// check version, enforces scheduled hard fork
	if !consensus::valid_header_version(header.height, header.version) {
		error!(
			"Invalid block header version received ({}), maybe update Grin?",
			header.version
		);
		return Err(ErrorKind::InvalidBlockVersion(header.version).into());
	}

	// TODO: remove CI check from here somehow
	if header.timestamp > Utc::now() + Duration::seconds(12 * (consensus::BLOCK_TIME_SEC as i64))
		&& !global::is_automated_testing_mode()
	{
		// refuse blocks more than 12 blocks intervals in future (as in bitcoin)
		// TODO add warning in p2p code if local time is too different from peers
		return Err(ErrorKind::InvalidBlockTime.into());
	}

	if !ctx.opts.contains(Options::SKIP_POW) {
		if !header.pow.is_primary() && !header.pow.is_secondary() {
			return Err(ErrorKind::LowEdgebits.into());
		}
		let edge_bits = header.pow.edge_bits();
		if !(ctx.pow_verifier)(header, edge_bits).is_ok() {
			error!(
				"pipe: error validating header with cuckoo edge_bits {}",
				edge_bits
			);
			return Err(ErrorKind::InvalidPow.into());
		}
	}

	// first I/O cost, better as late as possible
	let prev = match ctx.batch.get_previous_header(&header) {
		Ok(prev) => prev,
		Err(grin_store::Error::NotFoundErr(_)) => return Err(ErrorKind::Orphan.into()),
		Err(e) => {
			return Err(ErrorKind::StoreErr(
				e,
				format!("Failed to find previous header to {}", header.hash()),
			).into())
		}
	};

	// make sure this header has a height exactly one higher than the previous
	// header
	if header.height != prev.height + 1 {
		return Err(ErrorKind::InvalidBlockHeight.into());
	}

	// TODO - get rid of the automated testing mode check here somehow
	if header.timestamp <= prev.timestamp && !global::is_automated_testing_mode() {
		// prevent time warp attacks and some timestamp manipulations by forcing strict
		// time progression (but not in CI mode)
		return Err(ErrorKind::InvalidBlockTime.into());
	}

	// verify the proof of work and related parameters
	// at this point we have a previous block header
	// we know the height increased by one
	// so now we can check the total_difficulty increase is also valid
	// check the pow hash shows a difficulty at least as large
	// as the target difficulty
	if !ctx.opts.contains(Options::SKIP_POW) {
		if header.total_difficulty() <= prev.total_difficulty() {
			return Err(ErrorKind::DifficultyTooLow.into());
		}

		let target_difficulty = header.total_difficulty() - prev.total_difficulty();

		if header.pow.to_difficulty() < target_difficulty {
			return Err(ErrorKind::DifficultyTooLow.into());
		}

		// explicit check to ensure total_difficulty has increased by exactly
		// the _network_ difficulty of the previous block
		// (during testnet1 we use _block_ difficulty here)
		let prev = ctx.batch.get_previous_header(&header)?;
		let child_batch = ctx.batch.child()?;
		let diff_iter = store::DifficultyIter::from_batch(prev.hash(), child_batch);
		let next_header_info = consensus::next_difficulty(header.height, diff_iter);
		if target_difficulty != next_header_info.difficulty {
			info!(
				"validate_header: header target difficulty {} != {}",
				target_difficulty.to_num(),
				next_header_info.difficulty.to_num()
			);
			return Err(ErrorKind::WrongTotalDifficulty.into());
		}
		// check the secondary PoW scaling factor if applicable
		if header.pow.secondary_scaling != next_header_info.secondary_scaling {
			info!(
				"validate_header: header secondary scaling {} != {}",
				header.pow.secondary_scaling, next_header_info.secondary_scaling
			);
			return Err(ErrorKind::InvalidScaling.into());
		}
	}

	Ok(())
}

fn validate_block(block: &Block, ctx: &mut BlockContext) -> Result<(), Error> {
	let prev = ctx.batch.get_previous_header(&block.header)?;
	block
		.validate(&prev.total_kernel_offset, ctx.verifier_cache.clone())
		.map_err(|e| ErrorKind::InvalidBlockProof(e))?;
	Ok(())
}

/// TODO - This can move into the utxo_view.
/// Verify the block is not attempting to spend coinbase outputs
/// before they have sufficiently matured.
/// Note: requires a txhashset extension.
fn verify_coinbase_maturity(block: &Block, ext: &mut txhashset::Extension) -> Result<(), Error> {
	ext.verify_coinbase_maturity(&block.inputs(), block.header.height)?;
	Ok(())
}

/// Some "real magick" verification logic.
/// The (BlockSums, Block) tuple implements Committed...
/// This allows us to verify kernel sums across the full utxo and kernel sets
/// based on block_sums of previous block, accounting for the inputs|outputs|kernels
/// of the new block.
fn verify_block_sums(b: &Block, ext: &mut txhashset::Extension) -> Result<(), Error> {
	// TODO - this is 2 db calls, can we optimize this?
	// Retrieve the block_sums for the previous block.
	let prev = ext.batch.get_previous_header(&b.header)?;
	let block_sums = ext.batch.get_block_sums(&prev.hash())?;

	// Overage is based purely on the new block.
	// Previous block_sums have taken all previous overage into account.
	let overage = b.header.overage();

	// Offset on the other hand is the total kernel offset from the new block.
	let offset = b.header.total_kernel_offset();

	// Verify the kernel sums for the block_sums with the new block applied.
	let (utxo_sum, kernel_sum) =
		(block_sums, b as &Committed).verify_kernel_sums(overage, offset)?;

	// Save the new block_sums for the new block to the db via the batch.
	ext.batch.save_block_sums(
		&b.header.hash(),
		&BlockSums {
			utxo_sum,
			kernel_sum,
		},
	)?;

	Ok(())
}

/// Fully validate the block by applying it to the txhashset extension.
/// Check both the txhashset roots and sizes are correct after applying the block.
fn apply_block_to_txhashset(block: &Block, ext: &mut txhashset::Extension) -> Result<(), Error> {
	ext.validate_header_root(&block.header)?;
	ext.apply_block(block)?;
	ext.validate_roots()?;
	ext.validate_sizes()?;
	Ok(())
}

/// Officially adds the block to our chain.
/// Header must be added separately (assume this has been done previously).
fn add_block(b: &Block, batch: &store::Batch) -> Result<(), Error> {
	batch
		.save_block(b)
		.map_err(|e| ErrorKind::StoreErr(e, "pipe save block".to_owned()))?;
	Ok(())
}

/// Officially adds the block header to our header chain.
fn add_block_header(bh: &BlockHeader, batch: &store::Batch) -> Result<(), Error> {
	batch
		.save_block_header(bh)
		.map_err(|e| ErrorKind::StoreErr(e, "pipe save header".to_owned()))?;
	Ok(())
}

/// Directly updates the head if we've just appended a new block to it or handle
/// the situation where we've just added enough work to have a fork with more
/// work than the head.
fn update_head(b: &Block, ctx: &BlockContext) -> Result<Option<Tip>, Error> {
	// if we made a fork with more work than the head (which should also be true
	// when extending the head), update it
	let head = ctx.batch.head()?;
	if has_more_work(&b.header, &head) {
		// Update the block height index based on this new head.
		ctx.batch
			.setup_height(&b.header, &head)
			.map_err(|e| ErrorKind::StoreErr(e, "pipe setup height".to_owned()))?;

		let tip = Tip::from_header(&b.header);

		ctx.batch
			.save_body_head(&tip)
			.map_err(|e| ErrorKind::StoreErr(e, "pipe save body".to_owned()))?;

		debug!(
			"pipe: head updated to {} at {}",
			tip.last_block_h, tip.height
		);

		Ok(Some(tip))
	} else {
		Ok(None)
	}
}

// Whether the provided block totals more work than the chain tip
fn has_more_work(header: &BlockHeader, head: &Tip) -> bool {
	header.total_difficulty() > head.total_difficulty
}

/// Update the sync head so we can keep syncing from where we left off.
fn update_sync_head(bh: &BlockHeader, batch: &mut store::Batch) -> Result<(), Error> {
	let tip = Tip::from_header(bh);
	batch
		.save_sync_head(&tip)
		.map_err(|e| ErrorKind::StoreErr(e, "pipe save sync head".to_owned()))?;
	debug!("sync head {} @ {}", bh.hash(), bh.height);
	Ok(())
}

/// Update the header head if this header has most work.
fn update_header_head(bh: &BlockHeader, ctx: &mut BlockContext) -> Result<Option<Tip>, Error> {
	let header_head = ctx.batch.header_head()?;
	if has_more_work(&bh, &header_head) {
		let tip = Tip::from_header(bh);
		ctx.batch
			.save_header_head(&tip)
			.map_err(|e| ErrorKind::StoreErr(e, "pipe save header head".to_owned()))?;

		debug!(
			"pipe: header_head updated to {} at {}",
			tip.last_block_h, tip.height
		);

		Ok(Some(tip))
	} else {
		Ok(None)
	}
}

/// Rewind the header chain and reapply headers on a fork.
pub fn rewind_and_apply_header_fork(
	header: &BlockHeader,
	ext: &mut txhashset::HeaderExtension,
) -> Result<(), Error> {
	let mut fork_hashes = vec![];
	let mut current = ext.batch.get_previous_header(header)?;
	while current.height > 0 && !ext.batch.is_on_current_chain(&current).is_ok() {
		fork_hashes.push(current.hash());
		current = ext.batch.get_previous_header(&current)?;
	}
	fork_hashes.reverse();

	let forked_header = current;

	// Rewind the txhashset state back to the block where we forked from the most work chain.
	ext.rewind(&forked_header)?;

	// Re-apply all headers on this fork.
	for h in fork_hashes {
		let header = ext
			.batch
			.get_block_header(&h)
			.map_err(|e| ErrorKind::StoreErr(e, format!("getting forked headers")))?;
		ext.apply_header(&header)?;
	}
	Ok(())
}

/// Utility function to handle forks. From the forked block, jump backward
/// to find to fork root. Rewind the txhashset to the root and apply all the
/// forked blocks prior to the one being processed to set the txhashset in
/// the expected state.
pub fn rewind_and_apply_fork(b: &Block, ext: &mut txhashset::Extension) -> Result<(), Error> {
	// extending a fork, first identify the block where forking occurred
	// keeping the hashes of blocks along the fork
	let mut fork_hashes = vec![];
	let mut current = ext.batch.get_previous_header(&b.header)?;
	while current.height > 0 && !ext.batch.is_on_current_chain(&current).is_ok() {
		fork_hashes.push(current.hash());
		current = ext.batch.get_previous_header(&current)?;
	}
	fork_hashes.reverse();

	let forked_header = current;

	// Rewind the txhashset state back to the block where we forked from the most work chain.
	ext.rewind(&forked_header)?;

	// Now re-apply all blocks on this fork.
	for h in fork_hashes {
		let fb = ext
			.batch
			.get_block(&h)
			.map_err(|e| ErrorKind::StoreErr(e, format!("getting forked blocks")))?;

		// Re-verify coinbase maturity along this fork.
		verify_coinbase_maturity(&fb, ext)?;
		// Validate the block against the UTXO set.
		validate_utxo(&fb, ext)?;
		// Re-verify block_sums to set the block_sums up on this fork correctly.
		verify_block_sums(&fb, ext)?;
		// Re-apply the blocks.
		apply_block_to_txhashset(&fb, ext)?;
	}
	Ok(())
}

fn validate_utxo(block: &Block, ext: &txhashset::Extension) -> Result<(), Error> {
	let utxo = ext.utxo_view();
	utxo.validate_block(block)?;
	Ok(())
}