refactor the state_sync to handle the long fork (#1902)

* split horizon into two explicit values for cut through and txhashset request

* let node which has 2-7 days of history be able to handle forks larger than 2 days

* add test simulate_long_fork

* add pause/resume feature on p2p for tests

* refactor the state_sync

* ignore the test case simulate_long_fork for normal Travis-CI

* refactor function check_txhashset_needed to be shared with body_sync

* fix: state TxHashsetDone should allow header sync

servers/src/common/types.rs

@@ -261,6 +261,8 @@ pub enum SyncStatus {
 	},
 	/// Finalizing the new state
 	TxHashsetSave,
+	/// State sync finalized
+	TxHashsetDone,
 	/// Downloading blocks
 	BodySync {
 		current_height: u64,
@@ -383,9 +385,6 @@ impl chain::TxHashsetWriteStatus for SyncState {
 	}
 
 	fn on_done(&self) {
-		self.update(SyncStatus::BodySync {
-			current_height: 0,
-			highest_height: 0,
-		});
+		self.update(SyncStatus::TxHashsetDone);
 	}
 }

servers/src/grin/seed.rs

@@ -41,6 +41,7 @@ pub fn connect_and_monitor(
 	seed_list: Box<Fn() -> Vec<SocketAddr> + Send>,
 	preferred_peers: Option<Vec<SocketAddr>>,
 	stop: Arc<AtomicBool>,
+	pause: Arc<AtomicBool>,
 ) {
 	let _ = thread::Builder::new()
 		.name("seed".to_string())
@@ -65,6 +66,12 @@ pub fn connect_and_monitor(
 			let mut start_attempt = 0;
 
 			while !stop.load(Ordering::Relaxed) {
+				// Pause egress peer connection request. Only for tests.
+				if pause.load(Ordering::Relaxed) {
+					thread::sleep(time::Duration::from_secs(1));
+					continue;
+				}
+
 				// Check for and remove expired peers from the storage
 				if Utc::now() - prev_expire_check > Duration::hours(1) {
 					peers.remove_expired();

servers/src/grin/server.rs

@@ -58,6 +58,8 @@ pub struct Server {
 	state_info: ServerStateInfo,
 	/// Stop flag
 	pub stop: Arc<AtomicBool>,
+	/// Pause flag
+	pub pause: Arc<AtomicBool>,
 }
 
 impl Server {
@@ -111,6 +113,7 @@ impl Server {
 		};
 
 		let stop = Arc::new(AtomicBool::new(false));
+		let pause = Arc::new(AtomicBool::new(false));
 
 		// Shared cache for verification results.
 		// We cache rangeproof verification and kernel signature verification.
@@ -173,6 +176,7 @@ impl Server {
 			net_adapter.clone(),
 			genesis.hash(),
 			stop.clone(),
+			pause.clone(),
 		)?);
 		chain_adapter.init(p2p_server.peers.clone());
 		pool_net_adapter.init(p2p_server.peers.clone());
@@ -203,6 +207,7 @@ impl Server {
 				seeder,
 				peers_preferred,
 				stop.clone(),
+				pause.clone(),
 			);
 		}
 
@@ -254,6 +259,7 @@ impl Server {
 				..Default::default()
 			},
 			stop,
+			pause,
 		})
 	}
 
@@ -425,6 +431,18 @@ impl Server {
 		self.stop.store(true, Ordering::Relaxed);
 	}
 
+	/// Pause the p2p server.
+	pub fn pause(&self) {
+		self.pause.store(true, Ordering::Relaxed);
+		thread::sleep(time::Duration::from_secs(1));
+		self.p2p.pause();
+	}
+
+	/// Resume the p2p server.
+	pub fn resume(&self) {
+		self.pause.store(false, Ordering::Relaxed);
+	}
+
 	/// Stops the test miner without stopping the p2p layer
 	pub fn stop_test_miner(&self, stop: Arc<AtomicBool>) {
 		stop.store(true, Ordering::Relaxed);

servers/src/grin/sync/body_sync.rs

@@ -19,8 +19,7 @@ use std::sync::Arc;
 
 use chain;
 use common::types::{SyncState, SyncStatus};
-use core::core::hash::Hashed;
-use core::global;
+use core::core::hash::Hash;
 use p2p;
 
 pub struct BodySync {
@@ -50,83 +49,39 @@ impl BodySync {
 		}
 	}
 
-	/// Check whether a body sync is needed and run it if so
+	/// Check whether a body sync is needed and run it if so.
+	/// Return true if txhashset download is needed (when requested block is under the horizon).
 	pub fn check_run(&mut self, head: &chain::Tip, highest_height: u64) -> bool {
-		// if fast_sync disabled or not needed, run the body_sync every 5s
+		// run the body_sync every 5s
 		if self.body_sync_due() {
-			self.body_sync();
+			if self.body_sync() {
+				return true;
+			}
 
 			self.sync_state.update(SyncStatus::BodySync {
 				current_height: head.height,
 				highest_height: highest_height,
 			});
-			return true;
 		}
 		false
 	}
 
-	fn body_sync(&mut self) {
-		let horizon = global::cut_through_horizon() as u64;
-		let body_head = self.chain.head().unwrap();
-		let header_head = self.chain.header_head().unwrap();
-		let sync_head = self.chain.get_sync_head().unwrap();
-
-		debug!(
-			"body_sync: body_head - {}, {}, header_head - {}, {}, sync_head - {}, {}",
-			body_head.last_block_h,
-			body_head.height,
-			header_head.last_block_h,
-			header_head.height,
-			sync_head.last_block_h,
-			sync_head.height,
-		);
-
-		let mut hashes = vec![];
-		let mut oldest_height = 0;
-
-		if header_head.total_difficulty > body_head.total_difficulty {
-			let mut current = self.chain.get_block_header(&header_head.last_block_h);
-
-			//+ remove me after #1880 root cause found
-			if current.is_err() {
-				error!(
-					"body_sync: header_head not found in chain db: {} at {}",
-					header_head.last_block_h, header_head.height,
-				);
-			}
-			//-
-
-			while let Ok(header) = current {
-				// break out of the while loop when we find a header common
-				// between the header chain and the current body chain
-				if let Ok(_) = self.chain.is_on_current_chain(&header) {
-					break;
-				}
-
-				hashes.push(header.hash());
-				oldest_height = header.height;
-				current = self.chain.get_previous_header(&header);
-			}
-		}
-		//+ remove me after #1880 root cause found
-		else {
+	/// Return true if txhashset download is needed (when requested block is under the horizon).
+	fn body_sync(&mut self) -> bool {
+		let mut hashes: Option<Vec<Hash>> = Some(vec![]);
+		if self
+			.chain
+			.check_txhashset_needed("body_sync".to_owned(), &mut hashes)
+		{
 			debug!(
-				"body_sync: header_head.total_difficulty: {}, body_head.total_difficulty: {}",
-				header_head.total_difficulty, body_head.total_difficulty,
+				"body_sync: cannot sync full blocks earlier than horizon. will request txhashset",
 			);
+			return true;
 		}
-		//-
 
+		let mut hashes = hashes.unwrap();
 		hashes.reverse();
 
-		if oldest_height < header_head.height.saturating_sub(horizon) {
-			debug!(
-				"body_sync: cannot sync full blocks earlier than horizon. oldest_height: {}",
-				oldest_height,
-			);
-			return;
-		}
-
 		let peers = self.peers.more_work_peers();
 
 		// if we have 5 peers to sync from then ask for 50 blocks total (peer_count *
@@ -147,6 +102,9 @@ impl BodySync {
 			.collect::<Vec<_>>();
 
 		if hashes_to_get.len() > 0 {
+			let body_head = self.chain.head().unwrap();
+			let header_head = self.chain.header_head().unwrap();
+
 			debug!(
 				"block_sync: {}/{} requesting blocks {:?} from {} peers",
 				body_head.height,
@@ -170,6 +128,7 @@ impl BodySync {
 				}
 			}
 		}
+		return false;
 	}
 
 	// Should we run block body sync and ask for more full blocks?

servers/src/grin/sync/header_sync.rs

@@ -51,7 +51,9 @@ impl HeaderSync {
 		}
 
 		let enable_header_sync = match self.sync_state.status() {
-			SyncStatus::BodySync { .. } | SyncStatus::HeaderSync { .. } => true,
+			SyncStatus::BodySync { .. }
+			| SyncStatus::HeaderSync { .. }
+			| SyncStatus::TxHashsetDone => true,
 			SyncStatus::NoSync | SyncStatus::Initial | SyncStatus::AwaitingPeers(_) => {
 				// Reset sync_head to header_head on transition to HeaderSync,
 				// but ONLY on initial transition to HeaderSync state.

servers/src/grin/sync/state_sync.rs

@@ -33,8 +33,8 @@ pub struct StateSync {
 	peers: Arc<p2p::Peers>,
 	chain: Arc<chain::Chain>,
 
-	prev_fast_sync: Option<DateTime<Utc>>,
-	fast_sync_peer: Option<Arc<Peer>>,
+	prev_state_sync: Option<DateTime<Utc>>,
+	state_sync_peer: Option<Arc<Peer>>,
 }
 
 impl StateSync {
@@ -47,8 +47,8 @@ impl StateSync {
 			sync_state,
 			peers,
 			chain,
-			prev_fast_sync: None,
-			fast_sync_peer: None,
+			prev_state_sync: None,
+			state_sync_peer: None,
 		}
 	}
 
@@ -59,13 +59,12 @@ impl StateSync {
 		&mut self,
 		header_head: &chain::Tip,
 		head: &chain::Tip,
+		tail: &chain::Tip,
 		highest_height: u64,
 	) -> bool {
-		let need_state_sync =
-			highest_height.saturating_sub(head.height) > global::cut_through_horizon() as u64;
-		if !need_state_sync {
-			return false;
-		}
+		trace!("state_sync: head.height: {}, tail.height: {}. header_head.height: {}, highest_height: {}",
+			   head.height, tail.height, header_head.height, highest_height,
+		);
 
 		let mut sync_need_restart = false;
 
@@ -73,47 +72,63 @@ impl StateSync {
 		{
 			let clone = self.sync_state.sync_error();
 			if let Some(ref sync_error) = *clone.read() {
-				error!("fast_sync: error = {:?}. restart fast sync", sync_error);
+				error!("state_sync: error = {:?}. restart fast sync", sync_error);
 				sync_need_restart = true;
 			}
 			drop(clone);
 		}
 
 		// check peer connection status of this sync
-		if let Some(ref peer) = self.fast_sync_peer {
+		if let Some(ref peer) = self.state_sync_peer {
 			if let SyncStatus::TxHashsetDownload { .. } = self.sync_state.status() {
 				if !peer.is_connected() {
 					sync_need_restart = true;
 					info!(
-						"fast_sync: peer connection lost: {:?}. restart",
+						"state_sync: peer connection lost: {:?}. restart",
 						peer.info.addr,
 					);
 				}
 			}
 		}
 
-		if sync_need_restart {
-			self.fast_sync_reset();
+		// if txhashset downloaded and validated successfully, we switch to BodySync state,
+		// and we need call state_sync_reset() to make it ready for next possible state sync.
+		let done = if let SyncStatus::TxHashsetDone = self.sync_state.status() {
+			self.sync_state.update(SyncStatus::BodySync {
+				current_height: 0,
+				highest_height: 0,
+			});
+			true
+		} else {
+			false
+		};
+
+		if sync_need_restart || done {
+			self.state_sync_reset();
 			self.sync_state.clear_sync_error();
 		}
 
+		if done {
+			return false;
+		}
+
 		// run fast sync if applicable, normally only run one-time, except restart in error
 		if header_head.height == highest_height {
-			let (go, download_timeout) = self.fast_sync_due();
+			let (go, download_timeout) = self.state_sync_due();
 
 			if let SyncStatus::TxHashsetDownload { .. } = self.sync_state.status() {
 				if download_timeout {
-					error!("fast_sync: TxHashsetDownload status timeout in 10 minutes!");
+					error!("state_sync: TxHashsetDownload status timeout in 10 minutes!");
 					self.sync_state
 						.set_sync_error(Error::P2P(p2p::Error::Timeout));
 				}
 			}
 
 			if go {
-				self.fast_sync_peer = None;
+				self.state_sync_peer = None;
 				match self.request_state(&header_head) {
 					Ok(peer) => {
-						self.fast_sync_peer = Some(peer);
+						self.state_sync_peer = Some(peer);
 					}
 					Err(e) => self.sync_state.set_sync_error(Error::P2P(e)),
 				}
@@ -141,28 +156,27 @@ impl StateSync {
 	}
 
 	fn request_state(&self, header_head: &chain::Tip) -> Result<Arc<Peer>, p2p::Error> {
-		let horizon = global::cut_through_horizon() as u64;
+		let threshold = global::state_sync_threshold() as u64;
 
 		if let Some(peer) = self.peers.most_work_peer() {
-			// ask for txhashset at 90% of horizon, this still leaves time for download
-			// and validation to happen and stay within horizon
+			// ask for txhashset at state_sync_threshold
 			let mut txhashset_head = self
 				.chain
 				.get_block_header(&header_head.prev_block_h)
 				.unwrap();
-			for _ in 0..(horizon - horizon / 10) {
+			for _ in 0..threshold {
 				txhashset_head = self.chain.get_previous_header(&txhashset_head).unwrap();
 			}
 			let bhash = txhashset_head.hash();
 			debug!(
-				"fast_sync: before txhashset request, header head: {} / {}, txhashset_head: {} / {}",
+				"state_sync: before txhashset request, header head: {} / {}, txhashset_head: {} / {}",
 				header_head.height,
 				header_head.last_block_h,
 				txhashset_head.height,
 				bhash
 			);
 			if let Err(e) = peer.send_txhashset_request(txhashset_head.height, bhash) {
-				error!("fast_sync: send_txhashset_request err! {:?}", e);
+				error!("state_sync: send_txhashset_request err! {:?}", e);
 				return Err(e);
 			}
 			return Ok(peer.clone());
@@ -171,13 +185,13 @@ impl StateSync {
 	}
 
 	// For now this is a one-time thing (it can be slow) at initial startup.
-	fn fast_sync_due(&mut self) -> (bool, bool) {
+	fn state_sync_due(&mut self) -> (bool, bool) {
 		let now = Utc::now();
 		let mut download_timeout = false;
 
-		match self.prev_fast_sync {
+		match self.prev_state_sync {
 			None => {
-				self.prev_fast_sync = Some(now);
+				self.prev_state_sync = Some(now);
 				(true, download_timeout)
 			}
 			Some(prev) => {
@@ -189,8 +203,8 @@ impl StateSync {
 		}
 	}
 
-	fn fast_sync_reset(&mut self) {
-		self.prev_fast_sync = None;
-		self.fast_sync_peer = None;
+	fn state_sync_reset(&mut self) {
+		self.prev_state_sync = None;
+		self.state_sync_peer = None;
 	}
 }

servers/src/grin/sync/syncer.rs

@@ -141,13 +141,34 @@ impl SyncRunner {
 
 			// if syncing is needed
 			let head = self.chain.head().unwrap();
+			let tail = self.chain.tail().unwrap_or_else(|_| head.clone());
 			let header_head = self.chain.header_head().unwrap();
 
 			// run each sync stage, each of them deciding whether they're needed
-			// except for body sync that only runs if state sync is off or done
+			// except for state sync that only runs if body sync return true (means txhashset is needed)
 			header_sync.check_run(&header_head, highest_height);
-			if !state_sync.check_run(&header_head, &head, highest_height) {
-				body_sync.check_run(&head, highest_height);
+
+			let mut check_state_sync = false;
+			match self.sync_state.status() {
+				SyncStatus::TxHashsetDownload { .. }
+				| SyncStatus::TxHashsetSetup
+				| SyncStatus::TxHashsetValidation { .. }
+				| SyncStatus::TxHashsetSave
+				| SyncStatus::TxHashsetDone => check_state_sync = true,
+				_ => {
+					// skip body sync if header chain is not synced.
+					if header_head.height < highest_height {
+						continue;
+					}
+
+					if body_sync.check_run(&head, highest_height) {
+						check_state_sync = true;
+					}
+				}
+			}
+
+			if check_state_sync {
+				state_sync.check_run(&header_head, &head, &tail, highest_height);
 			}
 		}
 	}