Lots of chain sync and block validation fixes

* Fix for the chain pipeline partly relying on an outdated head,
leading to not properly recognizing a fork and inconsistent sum
tree state.
* Do not drop block requests during sync that don't get satisfied,
retry enough time to get them and avoid stall.
* Always validate header, even in sync where we may have validated
it already. We don't want a block coming from a peer that could
squeeze through with an invalid header.
* When syncing, do not mark blocks that were errored by the chain
as received (typical case: orphan). Keep retrying.
* Improved chain state dump for debugging.
* Do not add to orphans blocks too far in the future.
* Better error reporting on db errors.
* Related sync test fixes.

TODO figure out why syncing peers timeout so often, very useful
to test but not that great for a fast sync experience.

chain/src/chain.rs

@@ -96,7 +96,7 @@ impl Chain {
 				info!(LOGGER, "Saved genesis block with hash {}", gen.hash());
 				tip
 			}
-			Err(e) => return Err(Error::StoreErr(e)),
+			Err(e) => return Err(Error::StoreErr(e, "chain init load head".to_owned())),
 		};
 
 		let store = Arc::new(chain_store);
@@ -116,7 +116,8 @@ impl Chain {
 	/// has been added to the longest chain, None if it's added to an (as of
 	/// now) orphan chain.
 	pub fn process_block(&self, b: Block, opts: Options) -> Result<Option<Tip>, Error> {
-		let head = self.store.head().map_err(&Error::StoreErr)?;
+		let head = self.store.head().map_err(|e| Error::StoreErr(e, "chain load head".to_owned()))?;
+    let height = head.height;
 		let ctx = self.ctx_from_head(head, opts);
 
 		let res = pipe::process_block(&b, ctx);
@@ -140,9 +141,11 @@ impl Chain {
 			}
 			Ok(None) => {}
 			Err(Error::Orphan) => {
-				let mut orphans = self.orphans.lock().unwrap();
-				orphans.push_front((opts, b));
-				orphans.truncate(MAX_ORPHANS);
+        if b.header.height < height + (MAX_ORPHANS as u64) {
+          let mut orphans = self.orphans.lock().unwrap();
+          orphans.push_front((opts, b));
+          orphans.truncate(MAX_ORPHANS);
+        }
 			}
 			Err(ref e) => {
 				info!(
@@ -166,7 +169,7 @@ impl Chain {
 		opts: Options,
 	) -> Result<Option<Tip>, Error> {
 
-		let head = self.store.get_header_head().map_err(&Error::StoreErr)?;
+		let head = self.store.get_header_head().map_err(|e| Error::StoreErr(e, "chain header head".to_owned()))?;
 		let ctx = self.ctx_from_head(head, opts);
 
 		pipe::process_block_header(bh, ctx)
@@ -221,8 +224,8 @@ impl Chain {
 		let sumtrees = self.sumtrees.read().unwrap();
 		let is_unspent = sumtrees.is_unspent(output_ref)?;
 		if is_unspent {
-			self.store.get_output_by_commit(output_ref).map_err(
-				&Error::StoreErr,
+			self.store.get_output_by_commit(output_ref).map_err(|e|
+			  Error::StoreErr(e, "chain get unspent".to_owned())
 			)
 		} else {
 			Err(Error::OutputNotFound)
@@ -259,23 +262,23 @@ impl Chain {
 
 	/// Block header for the chain head
 	pub fn head_header(&self) -> Result<BlockHeader, Error> {
-		self.store.head_header().map_err(&Error::StoreErr)
+		self.store.head_header().map_err(|e| Error::StoreErr(e, "chain head header".to_owned()))
 	}
 
 	/// Gets a block header by hash
 	pub fn get_block(&self, h: &Hash) -> Result<Block, Error> {
-		self.store.get_block(h).map_err(&Error::StoreErr)
+		self.store.get_block(h).map_err(|e| Error::StoreErr(e, "chain get block".to_owned()))
 	}
 
 	/// Gets a block header by hash
 	pub fn get_block_header(&self, h: &Hash) -> Result<BlockHeader, Error> {
-		self.store.get_block_header(h).map_err(&Error::StoreErr)
+		self.store.get_block_header(h).map_err(|e| Error::StoreErr(e, "chain get header".to_owned()))
 	}
 
 	/// Gets the block header at the provided height
 	pub fn get_header_by_height(&self, height: u64) -> Result<BlockHeader, Error> {
-		self.store.get_header_by_height(height).map_err(
-			&Error::StoreErr,
+		self.store.get_header_by_height(height).map_err(|e|
+			Error::StoreErr(e, "chain get header by height".to_owned()),
 		)
 	}
 
@@ -286,12 +289,12 @@ impl Chain {
 	) -> Result<BlockHeader, Error> {
 		self.store
 			.get_block_header_by_output_commit(commit)
-			.map_err(&Error::StoreErr)
+			.map_err(|e| Error::StoreErr(e, "chain get commitment".to_owned()))
 	}
 
 	/// Get the tip of the header chain
 	pub fn get_header_head(&self) -> Result<Tip, Error> {
-		self.store.get_header_head().map_err(&Error::StoreErr)
+		self.store.get_header_head().map_err(|e |Error::StoreErr(e, "chain get header head".to_owned()))
 	}
 
 	/// Builds an iterator on blocks starting from the current chain head and

chain/src/pipe.rs

@@ -63,10 +63,16 @@ pub fn process_block(b: &Block, mut ctx: BlockContext) -> Result<Option<Tip>, Er
 
 	validate_header(&b.header, &mut ctx)?;
 
-	// take the lock on the sum trees and start a chain extension unit of work
-	// dependent on the success of the internal validation and saving operations
+	// valid header, time to take the lock on the sum trees
 	let local_sumtrees = ctx.sumtrees.clone();
 	let mut sumtrees = local_sumtrees.write().unwrap();
+
+	// update head now that we're in the lock
+	ctx.head = ctx.store.head().
+		map_err(|e| Error::StoreErr(e, "pipe reload head".to_owned()))?;
+
+	// start a chain extension unit of work dependent on the success of the
+	// internal validation and saving operations
 	sumtree::extending(&mut sumtrees, |mut extension| {
 
 		validate_block(b, &mut ctx, &mut extension)?;
@@ -162,8 +168,8 @@ fn validate_header(header: &BlockHeader, ctx: &mut BlockContext) -> Result<(), E
 	}
 
 	// first I/O cost, better as late as possible
-	let prev = try!(ctx.store.get_block_header(&header.previous).map_err(
-		&Error::StoreErr,
+	let prev = try!(ctx.store.get_block_header(&header.previous).map_err(|e|
+		Error::StoreErr(e, format!("previous block header {}", header.previous)),
 	));
 
 	if header.height != prev.height + 1 {
@@ -208,19 +214,6 @@ fn validate_block(
 	let curve = secp::Secp256k1::with_caps(secp::ContextFlag::Commit);
 	try!(b.validate(&curve).map_err(&Error::InvalidBlockProof));
 
-	// check that all the outputs of the block are "new" -
-	// that they do not clobber any existing unspent outputs (by their commitment)
-	//
-	// TODO - do we need to do this here (and can we do this here if we need access
-	// to the chain)
-	// see check_duplicate_outputs in pool for the analogous operation on
-	// transaction outputs
-	// for output in &block.outputs {
-	// here we would check that the output is not a duplicate output based on the
-	// current chain
-	// };
-
-
 	// apply the new block to the MMR trees and check the new root hashes
 	if b.header.previous == ctx.head.last_block_h {
 		// standard head extension
@@ -268,7 +261,7 @@ fn validate_block(
 		kernel_root.hash != b.header.kernel_root
 	{
 
-		ext.dump();
+		ext.dump(false);
 		return Err(Error::InvalidRoot);
 	}
 
@@ -297,14 +290,12 @@ fn validate_block(
 
 /// Officially adds the block to our chain.
 fn add_block(b: &Block, ctx: &mut BlockContext) -> Result<(), Error> {
-	ctx.store.save_block(b).map_err(&Error::StoreErr)?;
-
-	Ok(())
+	ctx.store.save_block(b).map_err(|e| Error::StoreErr(e, "pipe save block".to_owned()))
 }
 
 /// Officially adds the block header to our header chain.
 fn add_block_header(bh: &BlockHeader, ctx: &mut BlockContext) -> Result<(), Error> {
-	ctx.store.save_block_header(bh).map_err(&Error::StoreErr)
+	ctx.store.save_block_header(bh).map_err(|e| Error::StoreErr(e, "pipe save header".to_owned()))
 }
 
 /// Directly updates the head if we've just appended a new block to it or handle
@@ -317,18 +308,16 @@ fn update_head(b: &Block, ctx: &mut BlockContext) -> Result<Option<Tip>, Error>
 	if tip.total_difficulty > ctx.head.total_difficulty {
 
 		// update the block height index
-		ctx.store.setup_height(&b.header).map_err(&Error::StoreErr)?;
+		ctx.store.setup_height(&b.header).map_err(|e| Error::StoreErr(e, "pipe setup height".to_owned()))?;
 
     // in sync mode, only update the "body chain", otherwise update both the
     // "header chain" and "body chain", updating the header chain in sync resets
 		// all additional "future" headers we've received
     if ctx.opts.intersects(SYNC) {
-      ctx.store.save_body_head(&tip).map_err(&Error::StoreErr)?;
+			ctx.store.save_body_head(&tip).map_err(|e| Error::StoreErr(e, "pipe save body".to_owned()))?;
     } else {
-      ctx.store.save_head(&tip).map_err(&Error::StoreErr)?;
+			ctx.store.save_head(&tip).map_err(|e| Error::StoreErr(e, "pipe save head".to_owned()))?;
     }      
-
-		ctx.store.save_head(&tip).map_err(&Error::StoreErr)?;
 		ctx.head = tip.clone();
 		info!(LOGGER, "Updated head to {} at {}.", b.hash(), b.header.height);
 		Ok(Some(tip))
@@ -345,7 +334,7 @@ fn update_header_head(bh: &BlockHeader, ctx: &mut BlockContext) -> Result<Option
 	// when extending the head), update it
 	let tip = Tip::from_block(bh);
 	if tip.total_difficulty > ctx.head.total_difficulty {
-		ctx.store.save_header_head(&tip).map_err(&Error::StoreErr)?;
+		ctx.store.save_header_head(&tip).map_err(|e| Error::StoreErr(e, "pipe save header head".to_owned()))?;
 
 		ctx.head = tip.clone();
 		info!(

chain/src/sumtree.rs

@@ -29,6 +29,7 @@ use grin_store;
 use grin_store::sumtree::PMMRBackend;
 use types::ChainStore;
 use types::Error;
+use util::LOGGER;
 
 const SUMTREES_SUBDIR: &'static str = "sumtrees";
 const UTXO_SUBDIR: &'static str = "utxo";
@@ -94,7 +95,7 @@ impl SumTrees {
 		match rpos {
 			Ok(pos) => Ok(self.output_pmmr_h.backend.get(pos).is_some()),
 			Err(grin_store::Error::NotFoundErr) => Ok(false),
-			Err(e) => Err(Error::StoreErr(e)),
+			Err(e) => Err(Error::StoreErr(e, "sumtree unspent check".to_owned())),
 		}
 	}
 }
@@ -115,6 +116,7 @@ where
 	let res: Result<T, Error>;
 	let rollback: bool;
 	{
+    debug!(LOGGER, "Starting new sumtree extension.");
 		let commit_index = trees.commit_index.clone();
 		let mut extension = Extension::new(trees, commit_index);
 		res = inner(&mut extension);
@@ -126,6 +128,7 @@ where
 	}
 	match res {
 		Err(e) => {
+      debug!(LOGGER, "Error returned, discarding sumtree extension.");
 			trees.output_pmmr_h.backend.discard();
 			trees.rproof_pmmr_h.backend.discard();
 			trees.kernel_pmmr_h.backend.discard();
@@ -133,10 +136,12 @@ where
 		}
 		Ok(r) => {
 			if rollback {
+        debug!(LOGGER, "Rollbacking sumtree extension.");
 				trees.output_pmmr_h.backend.discard();
 				trees.rproof_pmmr_h.backend.discard();
 				trees.kernel_pmmr_h.backend.discard();
 			} else {
+        debug!(LOGGER, "Committing sumtree extension.");
 				trees.output_pmmr_h.backend.sync()?;
 				trees.rproof_pmmr_h.backend.sync()?;
 				trees.kernel_pmmr_h.backend.sync()?;
@@ -145,6 +150,7 @@ where
 				trees.kernel_pmmr_h.last_pos = sizes.2;
 			}
 
+      debug!(LOGGER, "Sumtree extension done.");
 			Ok(r)
 		}
 	}
@@ -264,6 +270,7 @@ impl<'a> Extension<'a> {
 		let out_pos_rew = self.commit_index.get_output_pos(&output.commitment())?;
 		let kern_pos_rew = self.commit_index.get_kernel_pos(&kernel.excess)?;
 
+    debug!(LOGGER, "Rewind sumtrees to {}", out_pos_rew);
 		self.output_pmmr
 			.rewind(out_pos_rew, height as u32)
 			.map_err(&Error::SumTreeErr)?;
@@ -273,6 +280,7 @@ impl<'a> Extension<'a> {
 		self.kernel_pmmr
 			.rewind(kern_pos_rew, height as u32)
 			.map_err(&Error::SumTreeErr)?;
+    self.dump(true);
 		Ok(())
 	}
 
@@ -293,14 +301,17 @@ impl<'a> Extension<'a> {
 		self.rollback = true;
 	}
 
-	/// Dumps the state of the 3 sum trees to stdout for debugging
-	pub fn dump(&self) {
-		println!("-- outputs --");
-		self.output_pmmr.dump();
-		println!("-- range proofs --");
-		self.rproof_pmmr.dump();
-		println!("-- kernels --");
-		self.kernel_pmmr.dump();
+	/// Dumps the state of the 3 sum trees to stdout for debugging. Short version
+  /// only prints the UTXO tree.
+	pub fn dump(&self, short: bool) {
+		debug!(LOGGER, "-- outputs --");
+		self.output_pmmr.dump(short);
+    if !short {
+      debug!(LOGGER, "-- range proofs --");
+      self.rproof_pmmr.dump(short);
+      debug!(LOGGER, "-- kernels --");
+      self.kernel_pmmr.dump(short);
+    }
 	}
 
 	// Sizes of the sum trees, used by `extending` on rollback.

chain/src/types.rs

@@ -75,7 +75,7 @@ pub enum Error {
 	/// Invalid block version, either a mistake or outdated software
 	InvalidBlockVersion(u16),
 	/// Internal issue when trying to save or load data from store
-	StoreErr(grin_store::Error),
+	StoreErr(grin_store::Error, String),
 	/// Error serializing or deserializing a type
 	SerErr(ser::Error),
 	/// Error while updating the sum trees
@@ -88,7 +88,7 @@ pub enum Error {
 
 impl From<grin_store::Error> for Error {
 	fn from(e: grin_store::Error) -> Error {
-		Error::StoreErr(e)
+		Error::StoreErr(e, "wrapped".to_owned())
 	}
 }
 impl From<ser::Error> for Error {