Backporting to beta (#4203)

* Minor typo to ensure it updates only when synced. (#4188)

* Updater fixes (#4196)

* Minor typo to ensure it updates only when synced.

* Fix deadlock.

* Skip unneeded arg in making list.

* Allow auto-restart even when not running an update.

* Fix trace.

* Update update info on each loop.

* Fix build.

* Shutdown all sockets

* Remove superfluous use.

* Poll for upgrades as part of global status (long) (#4197)

* Poll for upgrades as part of global status (long)

* Fix path

* Prevent duplicate incoming connections (#4180)

js/src/modals/UpgradeParity/store.js

@@ -27,7 +27,7 @@ const STEP_UPDATING = 1;
 const STEP_COMPLETED = 2;
 const STEP_ERROR = 2;
 
-const CHECK_INTERVAL = 1 * A_MINUTE;
+let instance = null;
 
 export default class Store {
   @observable available = null;
@@ -44,8 +44,6 @@ export default class Store {
 
     this.loadStorage();
     this.checkUpgrade();
-
-    setInterval(this.checkUpgrade, CHECK_INTERVAL);
   }
 
   @computed get isVisible () {
@@ -119,10 +117,10 @@ export default class Store {
 
   checkUpgrade = () => {
     if (!this._api) {
-      return;
+      return Promise.resolve(false);
     }
 
-    Promise
+    return Promise
       .all([
         this._api.parity.upgradeReady(),
         this._api.parity.consensusCapability(),
@@ -134,11 +132,23 @@ export default class Store {
         }
 
         this.setVersions(available, version, consensusCapability);
+
+        return true;
       })
       .catch((error) => {
         console.warn('checkUpgrade', error);
+
+        return false;
       });
   }
+
+  static get (api) {
+    if (!instance) {
+      instance = new Store(api);
+    }
+
+    return instance;
+  }
 }
 
 export {

js/src/redux/providers/status.js

@@ -14,29 +14,30 @@
 // You should have received a copy of the GNU General Public License
 // along with Parity.  If not, see <http://www.gnu.org/licenses/>.
 
-import BalancesProvider from './balances';
-import { statusBlockNumber, statusCollection, statusLogs } from './statusActions';
 import { isEqual } from 'lodash';
 
 import { LOG_KEYS, getLogger } from '~/config';
+import UpgradeStore from '~/modals/UpgradeParity/store';
+
+import BalancesProvider from './balances';
+import { statusBlockNumber, statusCollection, statusLogs } from './statusActions';
 
 const log = getLogger(LOG_KEYS.Signer);
 let instance = null;
 
 export default class Status {
+  _apiStatus = {};
+  _status = {};
+  _longStatus = {};
+  _minerSettings = {};
+  _timeoutIds = {};
+  _blockNumberSubscriptionId = null;
+  _timestamp = Date.now();
+
   constructor (store, api) {
     this._api = api;
     this._store = store;
-
+    this._upgradeStore = UpgradeStore.get(api);
-    this._apiStatus = {};
-    this._status = {};
-    this._longStatus = {};
-    this._minerSettings = {};
-
-    this._timeoutIds = {};
-    this._blockNumberSubscriptionId = null;
-
-    this._timestamp = Date.now();
 
     // On connecting, stop all subscriptions
     api.on('connecting', this.stop, this);
@@ -281,10 +282,11 @@ export default class Status {
         this._api.parity.netChain(),
         this._api.parity.netPort(),
         this._api.parity.rpcSettings(),
-        this._api.parity.enode()
+        this._api.parity.enode(),
+        this._upgradeStore.checkUpgrade()
       ])
       .then(([
-        netPeers, clientVersion, netVersion, defaultExtraData, netChain, netPort, rpcSettings, enode
+        netPeers, clientVersion, netVersion, defaultExtraData, netChain, netPort, rpcSettings, enode, upgradeStatus
       ]) => {
         const isTest =
           netVersion === '2' || // morden

js/src/views/Application/application.js

@@ -51,7 +51,7 @@ class Application extends Component {
   }
 
   store = new Store(this.context.api);
-  upgradeStore = new UpgradeStore(this.context.api);
+  upgradeStore = UpgradeStore.get(this.context.api);
 
   render () {
     const [root] = (window.location.hash || '').replace('#/', '').split('/');
@@ -65,7 +65,11 @@ class Application extends Component {
 
     return (
       <div>
-        { isMinimized ? this.renderMinimized() : this.renderApp() }
+        {
+          isMinimized
+            ? this.renderMinimized()
+            : this.renderApp()
+        }
         <Connection />
         <ParityBar dapp={ isMinimized } />
       </div>

parity/main.rs

@@ -206,12 +206,27 @@ fn latest_exe_path() -> Option<PathBuf> {
 		.and_then(|mut f| { let mut exe = String::new(); f.read_to_string(&mut exe).ok().map(|_| updates_path(&exe)) })
 }
 
+#[cfg(windows)]
+fn global_cleanup() {
+	extern "system" { pub fn WSACleanup() -> i32; }
+	// We need to cleanup all sockets before spawning another Parity process. This makes shure everything is cleaned up.
+	// The loop is required because of internal refernce counter for winsock dll. We don't know how many crates we use do 
+	// initialize it. There's at least 2 now.
+	for _ in 0.. 10 {
+		unsafe { WSACleanup(); }
+	}
+}
+
+#[cfg(not(windows))]
+fn global_cleanup() {}
+
 // Starts ~/.parity-updates/parity and returns the code it exits with.
 fn run_parity() -> Option<i32> {
+	global_cleanup();
 	use ::std::ffi::OsString;
 	let prefix = vec![OsString::from("--can-restart"), OsString::from("--force-direct")];
 	latest_exe_path().and_then(|exe| process::Command::new(exe)
-		.args(&(env::args_os().chain(prefix.into_iter()).collect::<Vec<_>>()))
+		.args(&(env::args_os().skip(1).chain(prefix.into_iter()).collect::<Vec<_>>()))
 		.status()
 		.map(|es| es.code().unwrap_or(128))
 		.ok()
@@ -266,17 +281,23 @@ fn main() {
 	let exe = std::env::current_exe().ok();
 	let development = exe.as_ref().and_then(|p| p.parent().and_then(|p| p.parent()).and_then(|p| p.file_name()).map(|n| n == "target")).unwrap_or(false);
 	let same_name = exe.as_ref().map(|p| p.file_stem().map_or(false, |s| s == "parity") && p.extension().map_or(true, |x| x == "exe")).unwrap_or(false);
-	let latest_exe = latest_exe_path();
+	trace_main!("Starting up {} (force-direct: {}, development: {}, same-name: {})", std::env::current_exe().map(|x| format!("{}", x.display())).unwrap_or("<unknown>".to_owned()), force_direct, development, same_name);
-	let have_update = latest_exe.as_ref().map_or(false, |p| p.exists());
+	if !force_direct && !development && same_name {
-	let is_non_updated_current = exe.map_or(false, |exe| latest_exe.as_ref().map_or(false, |lexe| exe.canonicalize().ok() != lexe.canonicalize().ok()));
-	trace_main!("Starting up {} (force-direct: {}, development: {}, same-name: {}, have-update: {}, non-updated-current: {})", std::env::current_exe().map(|x| format!("{}", x.display())).unwrap_or("<unknown>".to_owned()), force_direct, development, same_name, have_update, is_non_updated_current);
-	if !force_direct && !development && same_name && have_update && is_non_updated_current {
 		// looks like we're not running ~/.parity-updates/parity when the user is expecting otherwise.
 		// Everything run inside a loop, so we'll be able to restart from the child into a new version seamlessly.
 		loop {
 			// If we fail to run the updated parity then fallback to local version.
-			trace_main!("Attempting to run latest update ({})...", latest_exe.as_ref().expect("guarded by have_update; latest_exe must exist for have_update; qed").display());
+			let latest_exe = latest_exe_path();
-			let exit_code = run_parity().unwrap_or_else(|| { trace_main!("Falling back to local..."); main_direct(true) });
+			let have_update = latest_exe.as_ref().map_or(false, |p| p.exists());
+			let is_non_updated_current = exe.as_ref().map_or(false, |exe| latest_exe.as_ref().map_or(false, |lexe| exe.canonicalize().ok() != lexe.canonicalize().ok()));
+			trace_main!("Starting... (have-update: {}, non-updated-current: {})", have_update, is_non_updated_current);
+			let exit_code = if have_update && is_non_updated_current {
+				trace_main!("Attempting to run latest update ({})...", latest_exe.as_ref().expect("guarded by have_update; latest_exe must exist for have_update; qed").display());
+				run_parity().unwrap_or_else(|| { trace_main!("Falling back to local..."); main_direct(true) })
+			} else {
+				trace_main!("No latest update. Attempting to direct...");
+				main_direct(true)
+			};
 			trace_main!("Latest exited with {}", exit_code);
 			if exit_code != PLEASE_RESTART_EXIT_CODE {
 				trace_main!("Quitting...");

updater/src/updater.rs

@@ -273,6 +273,7 @@ impl Updater {
 					if s.fetching.is_none() {
 						info!(target: "updater", "Attempting to get parity binary {}", b);
 						s.fetching = Some(latest.track.clone());
+						drop(s);
 						let weak_self = self.weak_self.lock().clone();
 						let f = move |r: Result<PathBuf, fetch::Error>| if let Some(this) = weak_self.upgrade() { this.fetch_done(r) };
 						self.fetcher.lock().as_ref().expect("Created on `new`; qed").fetch(b, Box::new(f));
@@ -311,7 +312,7 @@ impl Updater {
 impl ChainNotify for Updater {
 	fn new_blocks(&self, _imported: Vec<H256>, _invalid: Vec<H256>, _enacted: Vec<H256>, _retracted: Vec<H256>, _sealed: Vec<H256>, _proposed: Vec<Bytes>, _duration: u64) {
 		match (self.client.upgrade(), self.sync.upgrade()) {
-			(Some(ref c), Some(ref s)) if s.status().is_syncing(c.queue_info()) => self.poll(),
+			(Some(ref c), Some(ref s)) if !s.status().is_syncing(c.queue_info()) => self.poll(),
 			_ => {},
 		}
 	}

util/network/src/host.rs

@@ -683,8 +683,7 @@ impl Host {
 
 	#[cfg_attr(feature="dev", allow(single_match))]
 	fn connect_peer(&self, id: &NodeId, io: &IoContext<NetworkIoMessage>) {
-		if self.have_session(id)
+		if self.have_session(id) {
-		{
 			trace!(target: "network", "Aborted connect. Node already connected.");
 			return;
 		}
@@ -788,102 +787,119 @@ impl Host {
 		let mut packet_data: Vec<(ProtocolId, PacketId, Vec<u8>)> = Vec::new();
 		let mut kill = false;
 		let session = { self.sessions.read().get(token).cloned() };
+		let mut ready_id = None;
 		if let Some(session) = session.clone() {
-			let mut s = session.lock();
+			{
-			loop {
+				let mut s = session.lock();
-				let session_result = s.readable(io, &self.info.read());
+				loop {
-				match session_result {
+					let session_result = s.readable(io, &self.info.read());
-					Err(e) => {
+					match session_result {
-						trace!(target: "network", "Session read error: {}:{:?} ({:?}) {:?}", token, s.id(), s.remote_addr(), e);
+						Err(e) => {
-						if let NetworkError::Disconnect(DisconnectReason::IncompatibleProtocol) = e {
+							trace!(target: "network", "Session read error: {}:{:?} ({:?}) {:?}", token, s.id(), s.remote_addr(), e);
-							if let Some(id) = s.id() {
+							if let NetworkError::Disconnect(DisconnectReason::IncompatibleProtocol) = e {
-								if !self.reserved_nodes.read().contains(id) {
+								if let Some(id) = s.id() {
-									self.nodes.write().mark_as_useless(id);
+									if !self.reserved_nodes.read().contains(id) {
+										self.nodes.write().mark_as_useless(id);
+									}
 								}
 							}
-						}
+							kill = true;
-						kill = true;
+							break;
-						break;
+						},
-					},
+						Ok(SessionData::Ready) => {
-					Ok(SessionData::Ready) => {
+							self.num_sessions.fetch_add(1, AtomicOrdering::SeqCst);
-						self.num_sessions.fetch_add(1, AtomicOrdering::SeqCst);
+							let session_count = self.session_count();
-						let session_count = self.session_count();
+							let (min_peers, max_peers, reserved_only) = {
-						let (min_peers, max_peers, reserved_only) = {
+								let info = self.info.read();
-							let info = self.info.read();
+								let mut max_peers = info.config.max_peers;
-							let mut max_peers = info.config.max_peers;
+								for cap in s.info.capabilities.iter() {
-							for cap in s.info.capabilities.iter() {
+									if let Some(num) = info.config.reserved_protocols.get(&cap.protocol) {
-								if let Some(num) = info.config.reserved_protocols.get(&cap.protocol) {
+										max_peers += *num;
-									max_peers += *num;
+										break;
-									break;
+									}
+								}
+								(info.config.min_peers as usize, max_peers as usize, info.config.non_reserved_mode == NonReservedPeerMode::Deny)
+							};
+
+							let id = s.id().expect("Ready session always has id").clone();
+
+							// Check for the session limit. session_counts accounts for the new session.
+							if reserved_only ||
+								(s.info.originated && session_count > min_peers) ||
+								(!s.info.originated && session_count > max_peers) {
+								// only proceed if the connecting peer is reserved.
+								if !self.reserved_nodes.read().contains(&id) {
+									s.disconnect(io, DisconnectReason::TooManyPeers);
+									return;
 								}
 							}
-							(info.config.min_peers as usize, max_peers as usize, info.config.non_reserved_mode == NonReservedPeerMode::Deny)
+							ready_id = Some(id);
-						};
 
-						// Check for the session limit. session_counts accounts for the new session.
+							// Add it to the node table
-						if reserved_only ||
+							if !s.info.originated {
-							(s.info.originated && session_count > min_peers) ||
+								if let Ok(address) = s.remote_addr() {
-							(!s.info.originated && session_count > max_peers) {
+									let entry = NodeEntry { id: id, endpoint: NodeEndpoint { address: address, udp_port: address.port() } };
-							// only proceed if the connecting peer is reserved.
+									self.nodes.write().add_node(Node::new(entry.id.clone(), entry.endpoint.clone()));
-							if !self.reserved_nodes.read().contains(s.id().expect("Ready session always has id")) {
+									let mut discovery = self.discovery.lock();
-								s.disconnect(io, DisconnectReason::TooManyPeers);
+									if let Some(ref mut discovery) = *discovery {
-								return;
+										discovery.add_node(entry);
-							}
+									}
-						}
-
-						// Add it to the node table
-						if !s.info.originated {
-							if let Ok(address) = s.remote_addr() {
-								let entry = NodeEntry { id: s.id().expect("Ready session always has id").clone(), endpoint: NodeEndpoint { address: address, udp_port: address.port() } };
-								self.nodes.write().add_node(Node::new(entry.id.clone(), entry.endpoint.clone()));
-								let mut discovery = self.discovery.lock();
-								if let Some(ref mut discovery) = *discovery {
-									discovery.add_node(entry);
 								}
 							}
-						}
+							for (p, _) in self.handlers.read().iter() {
-						for (p, _) in self.handlers.read().iter() {
+								if s.have_capability(*p) {
-							if s.have_capability(*p) {
+									ready_data.push(*p);
-								ready_data.push(*p);
+								}
 							}
-						}
+						},
-					},
+						Ok(SessionData::Packet {
-					Ok(SessionData::Packet {
+							data,
-						data,
+							protocol,
-						protocol,
+							packet_id,
-						packet_id,
+						}) => {
-					}) => {
+							match self.handlers.read().get(&protocol) {
-						match self.handlers.read().get(&protocol) {
+								None => { warn!(target: "network", "No handler found for protocol: {:?}", protocol) },
-							None => { warn!(target: "network", "No handler found for protocol: {:?}", protocol) },
+								Some(_) => packet_data.push((protocol, packet_id, data)),
-							Some(_) => packet_data.push((protocol, packet_id, data)),
+							}
-						}
+						},
-					},
+						Ok(SessionData::Continue) => (),
-					Ok(SessionData::Continue) => (),
+						Ok(SessionData::None) => break,
-					Ok(SessionData::None) => break,
+					}
-				}
-            }
-		}
-		if kill {
-			self.kill_connection(token, io, true);
-		}
-		let handlers = self.handlers.read();
-		for p in ready_data {
-			self.stats.inc_sessions();
-			let reserved = self.reserved_nodes.read();
-			if let Some(h) = handlers.get(&p).clone() {
-				h.connected(&NetworkContext::new(io, p, session.clone(), self.sessions.clone(), &reserved), &token);
-
-				// accumulate pending packets.
-				if let Some(session) = session.as_ref() {
-					let mut session = session.lock();
-					packet_data.extend(session.mark_connected(p));
 				}
 			}
-		}
+
-		for (p, packet_id, data) in packet_data {
+			if kill {
-			let reserved = self.reserved_nodes.read();
+				self.kill_connection(token, io, true);
-			if let Some(h) = handlers.get(&p).clone() {
+			}
-				h.read(&NetworkContext::new(io, p, session.clone(), self.sessions.clone(), &reserved), &token, packet_id, &data[1..]);
+
+			let handlers = self.handlers.read();
+			if !ready_data.is_empty() {
+				let duplicate = self.sessions.read().iter().any(|e| {
+					let session = e.lock();
+					session.token() != token && session.info.id == ready_id
+				});
+				if duplicate {
+					trace!(target: "network", "Rejected duplicate connection: {}", token);
+					session.lock().disconnect(io, DisconnectReason::DuplicatePeer);
+					return;
+				}
+				for p in ready_data {
+					self.stats.inc_sessions();
+					let reserved = self.reserved_nodes.read();
+					if let Some(h) = handlers.get(&p).clone() {
+						h.connected(&NetworkContext::new(io, p, Some(session.clone()), self.sessions.clone(), &reserved), &token);
+						// accumulate pending packets.
+						let mut session = session.lock();
+						packet_data.extend(session.mark_connected(p));
+					}
+				}
+			}
+
+			for (p, packet_id, data) in packet_data {
+				let reserved = self.reserved_nodes.read();
+				if let Some(h) = handlers.get(&p).clone() {
+					h.read(&NetworkContext::new(io, p, Some(session.clone()), self.sessions.clone(), &reserved), &token, packet_id, &data[1..]);
+				}
 			}
 		}
 	}