Move SyncStatus and SyncState to chain crate (#2964)

* Moved SyncState and SyncStatus to chain

* Rustfmt

* Fix typo in SyncError

Cargo.toml

@@ -34,6 +34,7 @@ failure_derive = "0.1"
 
 grin_api = { path = "./api", version = "2.0.1-beta.1" }
 grin_config = { path = "./config", version = "2.0.1-beta.1" }
+grin_chain = { path = "./chain", version = "2.0.1-beta.1" }
 grin_core = { path = "./core", version = "2.0.1-beta.1" }
 grin_keychain = { path = "./keychain", version = "2.0.1-beta.1" }
 grin_p2p = { path = "./p2p", version = "2.0.1-beta.1" }

chain/src/error.rs

@@ -140,6 +140,9 @@ pub enum ErrorKind {
 	/// Internal Roaring Bitmap error
 	#[fail(display = "Roaring Bitmap error")]
 	Bitmap,
+	/// Error during chain sync
+	#[fail(display = "Sync error")]
+	SyncError(String),
 }
 
 impl Display for Error {

chain/src/lib.rs

@@ -45,4 +45,6 @@ pub mod types;
 pub use crate::chain::{Chain, MAX_ORPHAN_SIZE};
 pub use crate::error::{Error, ErrorKind};
 pub use crate::store::ChainStore;
-pub use crate::types::{BlockStatus, ChainAdapter, Options, Tip, TxHashsetWriteStatus};
+pub use crate::types::{
+	BlockStatus, ChainAdapter, Options, SyncState, SyncStatus, Tip, TxHashsetWriteStatus,
+};

chain/src/types.rs

@@ -14,10 +14,15 @@
 
 //! Base types that the block chain pipeline requires.
 
+use chrono::prelude::{DateTime, Utc};
+use std::sync::Arc;
+
 use crate::core::core::hash::{Hash, Hashed, ZERO_HASH};
 use crate::core::core::{Block, BlockHeader};
 use crate::core::pow::Difficulty;
 use crate::core::ser;
+use crate::error::Error;
+use crate::util::RwLock;
 
 bitflags! {
 /// Options for block validation
@@ -33,6 +38,171 @@ bitflags! {
 	}
 }
 
+/// Various status sync can be in, whether it's fast sync or archival.
+#[derive(Debug, Clone, Copy, Eq, PartialEq)]
+#[allow(missing_docs)]
+pub enum SyncStatus {
+	/// Initial State (we do not yet know if we are/should be syncing)
+	Initial,
+	/// Not syncing
+	NoSync,
+	/// Not enough peers to do anything yet, boolean indicates whether
+	/// we should wait at all or ignore and start ASAP
+	AwaitingPeers(bool),
+	/// Downloading block headers
+	HeaderSync {
+		current_height: u64,
+		highest_height: u64,
+	},
+	/// Downloading the various txhashsets
+	TxHashsetDownload {
+		start_time: DateTime<Utc>,
+		prev_update_time: DateTime<Utc>,
+		update_time: DateTime<Utc>,
+		prev_downloaded_size: u64,
+		downloaded_size: u64,
+		total_size: u64,
+	},
+	/// Setting up before validation
+	TxHashsetSetup,
+	/// Validating the full state
+	TxHashsetValidation {
+		kernels: u64,
+		kernel_total: u64,
+		rproofs: u64,
+		rproof_total: u64,
+	},
+	/// Finalizing the new state
+	TxHashsetSave,
+	/// State sync finalized
+	TxHashsetDone,
+	/// Downloading blocks
+	BodySync {
+		current_height: u64,
+		highest_height: u64,
+	},
+	Shutdown,
+}
+
+/// Current sync state. Encapsulates the current SyncStatus.
+pub struct SyncState {
+	current: RwLock<SyncStatus>,
+	sync_error: Arc<RwLock<Option<Error>>>,
+}
+
+impl SyncState {
+	/// Return a new SyncState initialize to NoSync
+	pub fn new() -> SyncState {
+		SyncState {
+			current: RwLock::new(SyncStatus::Initial),
+			sync_error: Arc::new(RwLock::new(None)),
+		}
+	}
+
+	/// Whether the current state matches any active syncing operation.
+	/// Note: This includes our "initial" state.
+	pub fn is_syncing(&self) -> bool {
+		*self.current.read() != SyncStatus::NoSync
+	}
+
+	/// Current syncing status
+	pub fn status(&self) -> SyncStatus {
+		*self.current.read()
+	}
+
+	/// Update the syncing status
+	pub fn update(&self, new_status: SyncStatus) {
+		if self.status() == new_status {
+			return;
+		}
+
+		let mut status = self.current.write();
+
+		debug!("sync_state: sync_status: {:?} -> {:?}", *status, new_status,);
+
+		*status = new_status;
+	}
+
+	/// Update txhashset downloading progress
+	pub fn update_txhashset_download(&self, new_status: SyncStatus) -> bool {
+		if let SyncStatus::TxHashsetDownload { .. } = new_status {
+			let mut status = self.current.write();
+			*status = new_status;
+			true
+		} else {
+			false
+		}
+	}
+
+	/// Communicate sync error
+	pub fn set_sync_error(&self, error: Error) {
+		*self.sync_error.write() = Some(error);
+	}
+
+	/// Get sync error
+	pub fn sync_error(&self) -> Arc<RwLock<Option<Error>>> {
+		Arc::clone(&self.sync_error)
+	}
+
+	/// Clear sync error
+	pub fn clear_sync_error(&self) {
+		*self.sync_error.write() = None;
+	}
+}
+
+impl TxHashsetWriteStatus for SyncState {
+	fn on_setup(&self) {
+		self.update(SyncStatus::TxHashsetSetup);
+	}
+
+	fn on_validation(&self, vkernels: u64, vkernel_total: u64, vrproofs: u64, vrproof_total: u64) {
+		let mut status = self.current.write();
+		match *status {
+			SyncStatus::TxHashsetValidation {
+				kernels,
+				kernel_total,
+				rproofs,
+				rproof_total,
+			} => {
+				let ks = if vkernels > 0 { vkernels } else { kernels };
+				let kt = if vkernel_total > 0 {
+					vkernel_total
+				} else {
+					kernel_total
+				};
+				let rps = if vrproofs > 0 { vrproofs } else { rproofs };
+				let rpt = if vrproof_total > 0 {
+					vrproof_total
+				} else {
+					rproof_total
+				};
+				*status = SyncStatus::TxHashsetValidation {
+					kernels: ks,
+					kernel_total: kt,
+					rproofs: rps,
+					rproof_total: rpt,
+				};
+			}
+			_ => {
+				*status = SyncStatus::TxHashsetValidation {
+					kernels: 0,
+					kernel_total: 0,
+					rproofs: 0,
+					rproof_total: 0,
+				}
+			}
+		}
+	}
+
+	fn on_save(&self) {
+		self.update(SyncStatus::TxHashsetSave);
+	}
+
+	fn on_done(&self) {
+		self.update(SyncStatus::TxHashsetDone);
+	}
+}
+
 /// A helper to hold the roots of the txhashset in order to keep them
 /// readable.
 #[derive(Debug)]

servers/src/common/adapters.rs

@@ -23,11 +23,9 @@ use std::sync::{Arc, Weak};
 use std::thread;
 use std::time::Instant;
 
-use crate::chain::{self, BlockStatus, ChainAdapter, Options};
+use crate::chain::{self, BlockStatus, ChainAdapter, Options, SyncState, SyncStatus};
 use crate::common::hooks::{ChainEvents, NetEvents};
-use crate::common::types::{
-	self, ChainValidationMode, DandelionEpoch, ServerConfig, SyncState, SyncStatus,
-};
+use crate::common::types::{ChainValidationMode, DandelionEpoch, ServerConfig};
 use crate::core::core::hash::{Hash, Hashed};
 use crate::core::core::transaction::Transaction;
 use crate::core::core::verifier_cache::VerifierCache;
@@ -251,7 +249,11 @@ impl p2p::ChainAdapter for NetToChainAdapter {
 			.process_block_header(&bh, self.chain_opts(false));
 
 		if let Err(e) = res {
-			debug!("Block header {} refused by chain: {:?}", bh.hash(), e.kind());
+			debug!(
+				"Block header {} refused by chain: {:?}",
+				bh.hash(),
+				e.kind()
+			);
 			if e.is_bad_data() {
 				return Ok(false);
 			} else {
@@ -428,7 +430,7 @@ impl p2p::ChainAdapter for NetToChainAdapter {
 			error!("Failed to save txhashset archive: {}", e);
 
 			let is_good_data = !e.is_bad_data();
-			self.sync_state.set_sync_error(types::Error::Chain(e));
+			self.sync_state.set_sync_error(e);
 			Ok(is_good_data)
 		} else {
 			info!("Received valid txhashset data for {}.", h);

servers/src/common/stats.rs

@@ -26,7 +26,7 @@ use crate::core::ser::ProtocolVersion;
 use chrono::prelude::*;
 
 use crate::chain;
-use crate::common::types::SyncStatus;
+use crate::chain::SyncStatus;
 use crate::p2p;
 
 /// Server state info collection struct, to be passed around into internals