Using multiple NTP servers (#6173)

* Small improvements to time estimation.

* Allow multiple NTP servers to be used.

* Removing boxing.

* Be nice.

* Be nicer.

* Update list of servers and add reference.

dapps/src/api/api.rs

@@ -21,7 +21,7 @@ use hyper::method::Method;
 use hyper::status::StatusCode;
 
 use api::{response, types};
-use api::time::TimeChecker;
+use api::time::{TimeChecker, MAX_DRIFT};
 use apps::fetcher::Fetcher;
 use handlers::{self, extract_url};
 use endpoint::{Endpoint, Handler, EndpointPath};
@@ -122,7 +122,6 @@ impl RestApiRouter {
 
 			// Check time
 			let time = {
-				const MAX_DRIFT: i64 = 500;
 				let (status, message, details) = match time {
 					Ok(Ok(diff)) if diff < MAX_DRIFT && diff > -MAX_DRIFT => {
 						(HealthStatus::Ok, "".into(), diff)

dapps/src/api/time.rs

@@ -33,17 +33,21 @@
 
 use std::io;
 use std::{fmt, mem, time};
+use std::sync::atomic::{self, AtomicUsize};
 
 use std::collections::VecDeque;
 use futures::{self, Future, BoxFuture};
-use futures_cpupool::CpuPool;
+use futures::future::{self, IntoFuture};
+use futures_cpupool::{CpuPool, CpuFuture};
 use ntp;
 use time::{Duration, Timespec};
 use util::{Arc, RwLock};
 
 /// Time checker error.
 #[derive(Debug, Clone, PartialEq)]
 pub enum Error {
+	/// No servers are currently available for a query.
+	NoServersAvailable,
 	/// There was an error when trying to reach the NTP server.
 	Ntp(String),
 	/// IO error when reading NTP response.
@@ -55,6 +59,7 @@ impl fmt::Display for Error {
 		use self::Error::*;
 
 		match *self {
+			NoServersAvailable => write!(fmt, "No NTP servers available"),
 			Ntp(ref err) => write!(fmt, "NTP error: {}", err),
 			Io(ref err) => write!(fmt, "Connection Error: {}", err),
 		}
@@ -71,58 +76,123 @@ impl From<ntp::errors::Error> for Error {
 
 /// NTP time drift checker.
 pub trait Ntp {
+	/// Returned Future.
+	type Future: IntoFuture<Item=Duration, Error=Error>;
+
 	/// Returns the current time drift.
-	fn drift(&self) -> BoxFuture<Duration, Error>;
+	fn drift(&self) -> Self::Future;
+}
+
+const SERVER_MAX_POLL_INTERVAL_SECS: u64 = 60;
+#[derive(Debug)]
+struct Server {
+	pub address: String,
+	next_call: RwLock<time::Instant>,
+	failures: AtomicUsize,
+}
+
+impl Server {
+	pub fn is_available(&self) -> bool {
+		*self.next_call.read() < time::Instant::now()
+	}
+
+	pub fn report_success(&self) {
+		self.failures.store(0, atomic::Ordering::SeqCst);
+		self.update_next_call(1)
+	}
+
+	pub fn report_failure(&self) {
+		let errors = self.failures.fetch_add(1, atomic::Ordering::SeqCst);
+		self.update_next_call(1 << errors)
+	}
+
+	fn update_next_call(&self, delay: usize) {
+		*self.next_call.write() = time::Instant::now() + time::Duration::from_secs(delay as u64 * SERVER_MAX_POLL_INTERVAL_SECS);
+	}
+}
+
+impl<T: AsRef<str>> From<T> for Server {
+	fn from(t: T) -> Self {
+		Server {
+			address: t.as_ref().to_owned(),
+			next_call: RwLock::new(time::Instant::now()),
+			failures: Default::default(),
+		}
+	}
 }
 
 /// NTP client using the SNTP algorithm for calculating drift.
 #[derive(Clone)]
 pub struct SimpleNtp {
-	address: Arc<String>,
+	addresses: Vec<Arc<Server>>,
 	pool: CpuPool,
 }
 
 impl fmt::Debug for SimpleNtp {
 	fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-		write!(f, "Ntp {{ address: {} }}", self.address)
+		f
+			.debug_struct("SimpleNtp")
+			.field("addresses", &self.addresses)
+			.finish()
 	}
 }
 
 impl SimpleNtp {
-	fn new(address: &str, pool: CpuPool) -> SimpleNtp {
+	fn new<T: AsRef<str>>(addresses: &[T], pool: CpuPool) -> SimpleNtp {
 		SimpleNtp {
-			address: Arc::new(address.to_owned()),
+			addresses: addresses.iter().map(Server::from).map(Arc::new).collect(),
 			pool: pool,
 		}
 	}
 }
 
 impl Ntp for SimpleNtp {
-	fn drift(&self) -> BoxFuture<Duration, Error> {
-		let address = self.address.clone();
-		if &*address == "none" {
-			return futures::future::err(Error::Ntp("NTP server is not provided.".into())).boxed();
-		}
-
-		self.pool.spawn_fn(move || {
-			let packet = ntp::request(&*address)?;
-			let dest_time = ::time::now_utc().to_timespec();
-			let orig_time = Timespec::from(packet.orig_time);
-			let recv_time = Timespec::from(packet.recv_time);
-			let transmit_time = Timespec::from(packet.transmit_time);
-
-			let drift = ((recv_time - orig_time) + (transmit_time - dest_time)) / 2;
-
-			Ok(drift)
-		}).boxed()
+	type Future = future::Either<
+		CpuFuture<Duration, Error>,
+		future::FutureResult<Duration, Error>,
+	>;
+
+	fn drift(&self) -> Self::Future {
+		use self::future::Either::{A, B};
+
+		let server = self.addresses.iter().find(|server| server.is_available());
+		server.map(|server| {
+			let server = server.clone();
+			A(self.pool.spawn_fn(move || {
+				debug!(target: "dapps", "Fetching time from {}.", server.address);
+
+				match ntp::request(&server.address) {
+					Ok(packet) => {
+						let dest_time = ::time::now_utc().to_timespec();
+						let orig_time = Timespec::from(packet.orig_time);
+						let recv_time = Timespec::from(packet.recv_time);
+						let transmit_time = Timespec::from(packet.transmit_time);
+
+						let drift = ((recv_time - orig_time) + (transmit_time - dest_time)) / 2;
+
+						server.report_success();
+						Ok(drift)
+					},
+					Err(err) => {
+						server.report_failure();
+						Err(err.into())
+					},
+				}
+			}))
+		}).unwrap_or_else(|| B(future::err(Error::NoServersAvailable)))
 	}
 }
 
 // NOTE In a positive scenario first results will be seen after:
-// MAX_RESULTS * UPDATE_TIMEOUT_OK_SECS seconds.
-const MAX_RESULTS: usize = 7;
-const UPDATE_TIMEOUT_OK_SECS: u64 = 30;
-const UPDATE_TIMEOUT_ERR_SECS: u64 = 2;
+// MAX_RESULTS * UPDATE_TIMEOUT_INCOMPLETE_SECS seconds.
+const MAX_RESULTS: usize = 4;
+const UPDATE_TIMEOUT_OK_SECS: u64 = 6 * 60 * 60;
+const UPDATE_TIMEOUT_WARN_SECS: u64 = 15 * 60;
+const UPDATE_TIMEOUT_ERR_SECS: u64 = 60;
+const UPDATE_TIMEOUT_INCOMPLETE_SECS: u64 = 10;
+
+/// Maximal valid time drift.
+pub const MAX_DRIFT: i64 = 500;
 
 #[derive(Debug, Clone)]
 /// A time checker.
@@ -133,13 +203,13 @@ pub struct TimeChecker<N: Ntp = SimpleNtp> {
 
 impl TimeChecker<SimpleNtp> {
 	/// Creates new time checker given the NTP server address.
-	pub fn new(ntp_address: String, pool: CpuPool) -> Self {
+	pub fn new<T: AsRef<str>>(ntp_addresses: &[T], pool: CpuPool) -> Self {
 		let last_result = Arc::new(RwLock::new(
 			// Assume everything is ok at the very beginning.
 			(time::Instant::now(), vec![Ok(0)].into())
 		));
 
-		let ntp = SimpleNtp::new(&ntp_address, pool);
+		let ntp = SimpleNtp::new(ntp_addresses, pool);
 
 		TimeChecker {
 			ntp,
@@ -148,22 +218,34 @@ impl TimeChecker<SimpleNtp> {
 	}
 }
 
-impl<N: Ntp> TimeChecker<N> {
+impl<N: Ntp> TimeChecker<N> where <N::Future as IntoFuture>::Future: Send + 'static {
 	/// Updates the time
 	pub fn update(&self) -> BoxFuture<i64, Error> {
+		trace!(target: "dapps", "Updating time from NTP.");
 		let last_result = self.last_result.clone();
-		self.ntp.drift().then(move |res| {
+		self.ntp.drift().into_future().then(move |res| {
+			let res = res.map(|d| d.num_milliseconds());
+
+			if let Err(Error::NoServersAvailable) = res {
+				debug!(target: "dapps", "No NTP servers available. Selecting an older result.");
+				return select_result(last_result.read().1.iter());
+			}
+
+			// Update the results.
 			let mut results = mem::replace(&mut last_result.write().1, VecDeque::new());
+			let has_all_results = results.len() >= MAX_RESULTS;
 			let valid_till = time::Instant::now() + time::Duration::from_secs(
-				if res.is_ok() && results.len() == MAX_RESULTS {
-					UPDATE_TIMEOUT_OK_SECS
-				} else {
-					UPDATE_TIMEOUT_ERR_SECS
+				match res {
+					Ok(time) if has_all_results && time < MAX_DRIFT => UPDATE_TIMEOUT_OK_SECS,
+					Ok(_) if has_all_results => UPDATE_TIMEOUT_WARN_SECS,
+					Err(_) if has_all_results => UPDATE_TIMEOUT_ERR_SECS,
+					_ => UPDATE_TIMEOUT_INCOMPLETE_SECS,
 				}
 			);
 
+			trace!(target: "dapps", "New time drift received: {:?}", res);
 			// Push the result.
-			results.push_back(res.map(|d| d.num_milliseconds()));
+			results.push_back(res);
 			while results.len() > MAX_RESULTS {
 				results.pop_front();
 			}
@@ -208,7 +290,7 @@ mod tests {
 	use std::cell::{Cell, RefCell};
 	use std::time::Instant;
 	use time::Duration;
-	use futures::{self, BoxFuture, Future};
+	use futures::{future, Future};
 	use super::{Ntp, TimeChecker, Error};
 	use util::RwLock;
 
@@ -223,9 +305,11 @@ mod tests {
 	}
 
 	impl Ntp for FakeNtp {
-		fn drift(&self) -> BoxFuture<Duration, Error> {
+		type Future = future::FutureResult<Duration, Error>;
+
+		fn drift(&self) -> Self::Future {
 			self.1.set(self.1.get() + 1);
-			futures::future::ok(self.0.borrow_mut().pop().expect("Unexpected call to drift().")).boxed()
+			future::ok(self.0.borrow_mut().pop().expect("Unexpected call to drift()."))
 		}
 	}
 

dapps/src/lib.rs

@@ -130,7 +130,7 @@ impl Middleware {
 
 	/// Creates new middleware for UI server.
 	pub fn ui<F: Fetch>(
-		ntp_server: &str,
+		ntp_servers: &[String],
 		pool: CpuPool,
 		remote: Remote,
 		dapps_domain: &str,
@@ -146,7 +146,7 @@ impl Middleware {
 		).embeddable_on(None).allow_dapps(false));
 		let special = {
 			let mut special = special_endpoints(
-				ntp_server,
+				ntp_servers,
 				pool,
 				content_fetcher.clone(),
 				remote.clone(),
@@ -171,7 +171,7 @@ impl Middleware {
 
 	/// Creates new Dapps server middleware.
 	pub fn dapps<F: Fetch>(
-		ntp_server: &str,
+		ntp_servers: &[String],
 		pool: CpuPool,
 		remote: Remote,
 		ui_address: Option<(String, u16)>,
@@ -203,7 +203,7 @@ impl Middleware {
 
 		let special = {
 			let mut special = special_endpoints(
-				ntp_server,
+				ntp_servers,
 				pool,
 				content_fetcher.clone(),
 				remote.clone(),
@@ -237,8 +237,8 @@ impl http::RequestMiddleware for Middleware {
 	}
 }
 
-fn special_endpoints(
-	ntp_server: &str,
+fn special_endpoints<T: AsRef<str>>(
+	ntp_servers: &[T],
 	pool: CpuPool,
 	content_fetcher: Arc<apps::fetcher::Fetcher>,
 	remote: Remote,
@@ -250,7 +250,7 @@ fn special_endpoints(
 	special.insert(router::SpecialEndpoint::Api, Some(api::RestApi::new(
 		content_fetcher,
 		sync_status,
-		api::TimeChecker::new(ntp_server.into(), pool),
+		api::TimeChecker::new(ntp_servers, pool),
 		remote,
 	)));
 	special

dapps/src/tests/helpers/mod.rs

@@ -255,7 +255,7 @@ impl Server {
 		fetch: F,
 	) -> Result<Server, http::Error> {
 		let middleware = Middleware::dapps(
-			"pool.ntp.org:123",
+			&["0.pool.ntp.org:123".into(), "1.pool.ntp.org:123".into()],
 			CpuPool::new(4),
 			remote,
 			signer_address,

parity/cli/config.toml

@@ -78,7 +78,7 @@ disable_periodic = true
 jit = false
 
 [misc]
-ntp_server = "pool.ntp.org:123"
+ntp_servers = ["0.parity.pool.ntp.org:123"]
 logging = "own_tx=trace"
 log_file = "/var/log/parity.log"
 color = true

parity/cli/mod.rs

@@ -356,8 +356,8 @@ usage! {
 			or |c: &Config| otry!(c.vm).jit.clone(),
 
 		// -- Miscellaneous Options
-		flag_ntp_server: String = "none",
-			or |c: &Config| otry!(c.misc).ntp_server.clone(),
+		flag_ntp_servers: String = "0.parity.pool.ntp.org:123,1.parity.pool.ntp.org:123,2.parity.pool.ntp.org:123,3.parity.pool.ntp.org:123",
+			or |c: &Config| otry!(c.misc).ntp_servers.clone().map(|vec| vec.join(",")),
 		flag_logging: Option<String> = None,
 			or |c: &Config| otry!(c.misc).logging.clone().map(Some),
 		flag_log_file: Option<String> = None,
@@ -595,7 +595,7 @@ struct VM {
 
 #[derive(Default, Debug, PartialEq, Deserialize)]
 struct Misc {
-	ntp_server: Option<String>,
+	ntp_servers: Option<Vec<String>>,
 	logging: Option<String>,
 	log_file: Option<String>,
 	color: Option<bool>,
@@ -897,7 +897,7 @@ mod tests {
 			flag_dapps_apis_all: None,
 
 			// -- Miscellaneous Options
-			flag_ntp_server: "none".into(),
+			flag_ntp_servers: "0.parity.pool.ntp.org:123,1.parity.pool.ntp.org:123,2.parity.pool.ntp.org:123,3.parity.pool.ntp.org:123".into(),
 			flag_version: false,
 			flag_logging: Some("own_tx=trace".into()),
 			flag_log_file: Some("/var/log/parity.log".into()),
@@ -1075,7 +1075,7 @@ mod tests {
 				jit: Some(false),
 			}),
 			misc: Some(Misc {
-				ntp_server: Some("pool.ntp.org:123".into()),
+				ntp_servers: Some(vec!["0.parity.pool.ntp.org:123".into()]),
 				logging: Some("own_tx=trace".into()),
 				log_file: Some("/var/log/parity.log".into()),
 				color: Some(true),