diff --git a/chain/src/chain.rs b/chain/src/chain.rs index 6816ed55..0fbd0dfd 100644 --- a/chain/src/chain.rs +++ b/chain/src/chain.rs @@ -15,7 +15,7 @@ //! Facade and handler for the rest of the blockchain implementation //! and mostly the chain pipeline. -use std::collections::HashMap; +use std::collections::{HashMap, VecDeque}; use std::fs::File; use std::sync::{Arc, Mutex, RwLock}; use std::time::{Duration, Instant}; @@ -42,6 +42,9 @@ pub const MAX_ORPHAN_SIZE: usize = 200; /// When evicting, very old orphans are evicted first const MAX_ORPHAN_AGE_SECS: u64 = 300; +/// Number of recent hashes we keep to de-duplicate block or header sends +const HASHES_CACHE_SIZE: usize = 50; + #[derive(Debug, Clone)] struct Orphan { block: Block, @@ -130,8 +133,11 @@ pub struct Chain { head: Arc>, orphans: Arc, - txhashset_lock: Arc>, txhashset: Arc>, + // Recently processed blocks to avoid double-processing + block_hashes_cache: Arc>>, + // Recently processed headers to avoid double-processing + header_hashes_cache: Arc>>, // POW verification function pow_verifier: fn(&BlockHeader, u8) -> bool, @@ -177,9 +183,10 @@ impl Chain { adapter: adapter, head: Arc::new(Mutex::new(head)), orphans: Arc::new(OrphanBlockPool::new()), - txhashset_lock: Arc::new(Mutex::new(false)), txhashset: Arc::new(RwLock::new(txhashset)), pow_verifier: pow_verifier, + block_hashes_cache: Arc::new(RwLock::new(VecDeque::with_capacity(HASHES_CACHE_SIZE))), + header_hashes_cache: Arc::new(RwLock::new(VecDeque::with_capacity(HASHES_CACHE_SIZE))), }) } @@ -215,6 +222,12 @@ impl Chain { let mut ctx = self.ctx_from_head(head, opts)?; let res = pipe::process_block(&b, &mut ctx); + + { + let mut cache = self.block_hashes_cache.write().unwrap(); + cache.push_front(b.hash()); + cache.truncate(HASHES_CACHE_SIZE); + } match res { Ok(Some(ref tip)) => { @@ -226,29 +239,17 @@ impl Chain { } // notifying other parts of the system of the update - if !opts.contains(Options::SYNC) { - // broadcast the block - let adapter = self.adapter.clone(); - adapter.block_accepted(&b, opts); - } + self.adapter.block_accepted(&b, opts); + Ok((Some(tip.clone()), Some(b))) } Ok(None) => { // block got accepted but we did not extend the head // so its on a fork (or is the start of a new fork) // broadcast the block out so everyone knows about the fork - // - // TODO - This opens us to an amplification attack on blocks - // mined at a low difficulty. We should suppress really old blocks - // or less relevant blocks somehow. - // We should also probably consider banning nodes that send us really old - // blocks. - // - if !opts.contains(Options::SYNC) { // broadcast the block - let adapter = self.adapter.clone(); - adapter.block_accepted(&b, opts); - } + self.adapter.block_accepted(&b, opts); + Ok((None, Some(b))) } Err(e) => { @@ -304,7 +305,13 @@ impl Chain { pub fn process_block_header(&self, bh: &BlockHeader, opts: Options) -> Result<(), Error> { let header_head = self.get_header_head()?; let mut ctx = self.ctx_from_head(header_head, opts)?; - pipe::process_block_header(bh, &mut ctx) + let res = pipe::process_block_header(bh, &mut ctx); + { + let mut cache = self.header_hashes_cache.write().unwrap(); + cache.push_front(bh.hash()); + cache.truncate(HASHES_CACHE_SIZE); + } + res } /// Attempt to add a new header to the header chain. @@ -328,6 +335,8 @@ impl Chain { store: self.store.clone(), head: head, pow_verifier: self.pow_verifier, + block_hashes_cache: self.block_hashes_cache.clone(), + header_hashes_cache: self.header_hashes_cache.clone(), txhashset: self.txhashset.clone(), }) } @@ -529,7 +538,6 @@ impl Chain { where T: TxHashsetWriteStatus, { - let _ = self.txhashset_lock.lock().unwrap(); status.on_setup(); let head = self.head().unwrap(); let header_head = self.get_header_head().unwrap(); diff --git a/chain/src/error.rs b/chain/src/error.rs index 66626a3b..d9c774d1 100644 --- a/chain/src/error.rs +++ b/chain/src/error.rs @@ -51,6 +51,9 @@ pub enum ErrorKind { /// The proof of work is invalid #[fail(display = "Invalid PoW")] InvalidPow, + /// Peer abusively sending us an old block we already have + #[fail(display = "Invalid PoW")] + OldBlock, /// The block doesn't sum correctly or a tx signature is invalid #[fail(display = "Invalid Block Proof")] InvalidBlockProof(block::Error), diff --git a/chain/src/pipe.rs b/chain/src/pipe.rs index f844c2dc..9c9ae577 100644 --- a/chain/src/pipe.rs +++ b/chain/src/pipe.rs @@ -14,6 +14,7 @@ //! Implementation of the chain block acceptance (or refusal) pipeline. +use std::collections::VecDeque; use std::sync::{Arc, RwLock}; use time; @@ -45,6 +46,10 @@ pub struct BlockContext { pub pow_verifier: fn(&BlockHeader, u8) -> bool, /// MMR sum tree states pub txhashset: Arc>, + /// Recently processed blocks to avoid double-processing + pub block_hashes_cache: Arc>>, + /// Recently processed headers to avoid double-processing + pub header_hashes_cache: Arc>>, } /// Runs the block processing pipeline, including validation and finding a @@ -175,16 +180,12 @@ pub fn process_block_header(bh: &BlockHeader, ctx: &mut BlockContext) -> Result< /// 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(bh: Hash, ctx: &mut BlockContext) -> Result<(), Error> { - // TODO ring buffer of the last few blocks that came through here if bh == ctx.head.last_block_h || bh == ctx.head.prev_block_h { return Err(ErrorKind::Unfit("already known".to_string()).into()); } - if let Ok(h) = ctx.store.get_block_header(&bh) { - // there is a window where a block header can be saved but the chain head not - // updated yet, we plug that window here by re-accepting the block - if h.total_difficulty <= ctx.head.total_difficulty { - return Err(ErrorKind::Unfit("already in store".to_string()).into()); - } + let cache = ctx.block_hashes_cache.read().unwrap(); + if cache.contains(&bh) { + return Err(ErrorKind::Unfit("already known".to_string()).into()); } Ok(()) } @@ -192,16 +193,12 @@ fn check_header_known(bh: Hash, ctx: &mut BlockContext) -> Result<(), Error> { /// Quick in-memory check to fast-reject any block we've already handled /// recently. Keeps duplicates from the network in check. fn check_known(bh: Hash, ctx: &mut BlockContext) -> Result<(), Error> { - // TODO ring buffer of the last few blocks that came through here if bh == ctx.head.last_block_h || bh == ctx.head.prev_block_h { return Err(ErrorKind::Unfit("already known".to_string()).into()); } - if let Ok(b) = ctx.store.get_block(&bh) { - // there is a window where a block can be saved but the chain head not - // updated yet, we plug that window here by re-accepting the block - if b.header.total_difficulty <= ctx.head.total_difficulty { - return Err(ErrorKind::Unfit("already in store".to_string()).into()); - } + let cache = ctx.header_hashes_cache.read().unwrap(); + if cache.contains(&bh) { + return Err(ErrorKind::Unfit("already known".to_string()).into()); } Ok(()) } @@ -209,7 +206,6 @@ fn check_known(bh: Hash, ctx: &mut BlockContext) -> Result<(), Error> { /// 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. -/// TODO require only the block header (with length information) 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) { @@ -312,6 +308,13 @@ fn validate_header(header: &BlockHeader, ctx: &mut BlockContext) -> Result<(), E } fn validate_block(b: &Block, ctx: &mut BlockContext) -> Result<(), Error> { + if ctx.store.block_exists(&b.hash())? { + if b.header.height < ctx.head.height.saturating_sub(50) { + return Err(ErrorKind::OldBlock.into()); + } else { + return Err(ErrorKind::Unfit("already known".to_string()).into()); + } + } let prev = ctx.store.get_block_header(&b.header.previous)?; b.validate(&prev.total_kernel_offset, &prev.total_kernel_sum) .map_err(|e| ErrorKind::InvalidBlockProof(e))?; diff --git a/servers/src/common/adapters.rs b/servers/src/common/adapters.rs index 046e85e7..9db95953 100644 --- a/servers/src/common/adapters.rs +++ b/servers/src/common/adapters.rs @@ -25,7 +25,7 @@ use std::time::Instant; use chain::{self, ChainAdapter, Options, Tip}; use common::types::{ChainValidationMode, ServerConfig, SyncState, SyncStatus}; -use core::core; +use core::{core, global}; use core::core::block::BlockHeader; use core::core::hash::{Hash, Hashed}; use core::core::target::Difficulty; @@ -68,6 +68,11 @@ impl p2p::ChainAdapter for NetToChainAdapter { } fn transaction_received(&self, tx: core::Transaction, stem: bool) { + // nothing much we can do with a new transaction while syncing + if self.sync_state.is_syncing() { + return; + } + let source = pool::TxSource { debug_name: "p2p".to_string(), identifier: "?.?.?.?".to_string(), @@ -339,6 +344,11 @@ impl p2p::ChainAdapter for NetToChainAdapter { txhashset_data: File, _peer_addr: SocketAddr, ) -> bool { + // check status again after download, in case 2 txhashsets made it somehow + if self.sync_state.status() != SyncStatus::TxHashsetDownload { + return true; + } + if let Err(e) = w(&self.chain).txhashset_write(h, txhashset_data, self.sync_state.as_ref()) { @@ -416,9 +426,18 @@ impl NetToChainAdapter { // pushing the new block through the chain pipeline // remembering to reset the head if we have a bad block fn process_block(&self, b: core::Block, addr: SocketAddr) -> bool { + let chain = w(&self.chain); + if !self.archive_mode { + let head = chain.head().unwrap(); + // we have a fast sync'd node and are sent a block older than our horizon, + // only sync can do something with that + if b.header.height < head.height.saturating_sub(global::cut_through_horizon() as u64) { + return true; + } + } + let prev_hash = b.header.previous; let bhash = b.hash(); - let chain = w(&self.chain); match chain.process_block(b, self.chain_opts()) { Ok((tip, _)) => { self.validate_chain(bhash); @@ -574,12 +593,17 @@ impl NetToChainAdapter { /// blockchain accepted a new block, asking the pool to update its state and /// the network to broadcast the block pub struct ChainToPoolAndNetAdapter { + sync_state: Arc, tx_pool: Arc>>, peers: OneTime>, } impl ChainAdapter for ChainToPoolAndNetAdapter { fn block_accepted(&self, b: &core::Block, opts: Options) { + if self.sync_state.is_syncing() { + return; + } + debug!(LOGGER, "adapter: block_accepted: {:?}", b.hash()); if let Err(e) = self.tx_pool.write().unwrap().reconcile_block(b) { @@ -633,10 +657,12 @@ impl ChainAdapter for ChainToPoolAndNetAdapter { impl ChainToPoolAndNetAdapter { /// Construct a ChainToPoolAndNetAdapter instance. pub fn new( + sync_state: Arc, tx_pool: Arc>>, ) -> ChainToPoolAndNetAdapter { ChainToPoolAndNetAdapter { - tx_pool: tx_pool, + sync_state, + tx_pool, peers: OneTime::new(), } } diff --git a/servers/src/grin/server.rs b/servers/src/grin/server.rs index 17cf0216..83a06b0f 100644 --- a/servers/src/grin/server.rs +++ b/servers/src/grin/server.rs @@ -119,7 +119,12 @@ impl Server { pool_net_adapter.clone(), ))); - let chain_adapter = Arc::new(ChainToPoolAndNetAdapter::new(tx_pool.clone())); + let sync_state = Arc::new(SyncState::new()); + + let chain_adapter = Arc::new(ChainToPoolAndNetAdapter::new( + sync_state.clone(), + tx_pool.clone(), + )); let genesis = match config.chain_type { global::ChainTypes::Testnet1 => genesis::genesis_testnet1(), @@ -143,7 +148,6 @@ impl Server { pool_adapter.set_chain(Arc::downgrade(&shared_chain)); - let sync_state = Arc::new(SyncState::new()); let awaiting_peers = Arc::new(AtomicBool::new(false)); let net_adapter = Arc::new(NetToChainAdapter::new(