Fix onion service restarting issues. Callback for onion service status

Author: bay

p2p/src/listen.rs

@@ -21,20 +21,22 @@ use std::sync::Arc;
 use std::time::Duration;
 use std::{io, thread};
 
-pub fn listen<F, G, H>(
+pub fn listen<F, G, H, K>(
 	context_id: u32,
 	stop_state: Arc<StopState>,
 	tor_config: Option<TorConfig>,
 	listen_addr: Option<SocketAddr>, // Port in needed in not internal tor will be used
 	onion_expanded_key: Option<[u8; 64]>,
 	started_service_callback: Option<F>,
 	failed_service_callback: Option<G>,
+	service_status_callback: Option<K>,
 	handle_new_peer_callback: H,
 ) -> Result<(), Error>
 where
 	F: Fn(Option<String>),
 	G: Fn(&Error) -> bool, // return true if want to exit on falure
 	H: Fn(TcpDataStream, Option<PeerAddr>),
+	K: Fn(bool) + Send + 'static + std::marker::Sync, // return true if want to exit on falure
 {
 	let tor_config = tor_config.unwrap_or(TorConfig::no_tor_config());
 
@@ -49,10 +51,11 @@ where
 				onion_expanded_key,
 				started_service_callback,
 				failed_service_callback,
+				service_status_callback,
 				handle_new_peer_callback,
 			)
 		} else {
-			// Listening on extended Tor, listening on sockets
+			// Listening on external Tor, listening on sockets
 			let onion_address = tor_config.onion_address.clone().ok_or(Error::ConfigError(
 				"For tor external config, internal onion address is not specified.".into(),
 			))?;
@@ -67,6 +70,7 @@ where
 				Some(onion_address),
 				started_service_callback,
 				failed_service_callback,
+				service_status_callback,
 				handle_new_peer_callback,
 			)
 		}
@@ -79,23 +83,26 @@ where
 			None,
 			started_service_callback,
 			failed_service_callback,
+			service_status_callback,
 			handle_new_peer_callback,
 		)
 	}
 }
 
-fn listen_onion_service<F, G, H>(
+fn listen_onion_service<F, G, H, K>(
 	context_id: u32,
 	stop_state: Arc<StopState>,
 	onion_expanded_key: Option<[u8; 64]>,
 	started_service_callback: Option<F>,
 	failed_service_callback: Option<G>,
+	service_status_callback: Option<K>,
 	handle_new_peer_callback: H,
 ) -> Result<(), Error>
 where
 	F: Fn(Option<String>),
 	G: Fn(&Error) -> bool, // return true if want to exit on falure
 	H: Fn(TcpDataStream, Option<PeerAddr>),
+	K: Fn(bool) + Send + 'static + std::marker::Sync, // return true if want to exit on falure
 {
 	info!("Starting TOR (Arti) service...");
 
@@ -110,36 +117,45 @@ where
 		"mwc-node",
 		started_service_callback,
 		failed_service_callback,
+		service_status_callback,
 		handle_new_peer_callback,
 	)
 }
 
 /// Starts a new TCP server and listen to incoming connections. This is a
 /// blocking call until the TCP server stops.
-fn listen_socket<F, G, H>(
+fn listen_socket<F, G, H, K>(
 	stop_state: Arc<StopState>,
 	host: IpAddr,
 	port: u16,
 	onion_service_address: Option<String>,
 	started_service_callback: Option<F>,
 	failed_service_callback: Option<G>,
+	service_status_callback: Option<K>,
 	handle_new_peer_callback: H,
 ) -> Result<(), Error>
 where
 	F: Fn(Option<String>),
 	G: Fn(&Error) -> bool, // return true if want to exit on falure
 	H: Fn(TcpDataStream, Option<PeerAddr>),
+	K: Fn(bool) + Send + 'static + std::marker::Sync, // return true if want to exit on falure
 {
 	// start TCP listener and handle incoming connections
 	let addr = SocketAddr::new(host, port);
 	let listener = match run_global_async_block(async { TcpListener::bind(addr).await }) {
 		Ok(listener) => {
+			if let Some(f) = service_status_callback.as_ref() {
+				f(true)
+			}
 			if let Some(started_service_callback) = started_service_callback {
 				started_service_callback(onion_service_address);
 			}
 			listener
 		}
 		Err(e) => {
+			if let Some(f) = service_status_callback.as_ref() {
+				f(false)
+			}
 			let err = Error::TorProcess(format!(
 				"Unable to start listening on {}:{}, {}",
 				host, port, e
@@ -215,5 +231,9 @@ where
 		}
 		thread::sleep(Duration::from_millis(5));
 	}
+	if let Some(f) = service_status_callback.as_ref() {
+		f(false)
+	}
+
 	Ok(())
 }

p2p/src/serv.rs

@@ -193,6 +193,7 @@ impl Server {
 			onion_expanded_key,
 			Some(service_started_callback),
 			Some(service_failed_callback),
+			None::<Box<dyn Fn(bool) + Send + Sync + 'static>>,
 			handle_new_connection_callback,
 		)
 	}

p2p/src/tor/onion_service.rs

@@ -19,14 +19,16 @@ use crate::{Error, PeerAddr};
 use async_std::stream::StreamExt;
 use mwc_util::StopState;
 use std::pin::Pin;
-use std::sync::atomic::{AtomicI64, Ordering};
+use std::sync::atomic::{AtomicI64, AtomicU32, Ordering};
 use std::sync::Arc;
 use std::thread;
 use std::time::{Duration, Instant};
 use tor_cell::relaycell::msg::Connected;
 use tor_hsservice::status::State;
 use tor_proto::client::stream::IncomingStreamRequest;
 
+static SERVICE_COUNTER: AtomicU32 = AtomicU32::new(0);
+
 // started_service_callback accepting onion address
 fn start_arti<F>(
 	context_id: u32,
@@ -51,7 +53,12 @@ where
 	) = arti::access_arti(|tor_client| {
 		let (onion_service, onion_address, incoming_requests) = ArtiCore::start_onion_service(
 			&tor_client,
-			format!("onion-service-{}-{}", service_name, context_id),
+			format!(
+				"onion-service-{}-{}-{}",
+				service_name,
+				context_id,
+				SERVICE_COUNTER.fetch_add(1, Ordering::Relaxed)
+			),
 			onion_expanded_key.clone(),
 		)?;
 		Ok((
@@ -77,24 +84,31 @@ where
 /// Listening on Onion service.
 /// Here is we have a full workflow that monitors the onion service, restarting on network failre. It is
 /// Used at both mwc-node and mwc-wallet.
-pub fn listen_onion_service<F, G, H>(
+pub fn listen_onion_service<F, G, H, K>(
 	context_id: u32,
 	stop_state: Arc<StopState>,
 	onion_expanded_key: [u8; 64],
 	service_name: &str,
 	started_service_callback: Option<F>,
 	failed_service_callback: Option<G>,
+	service_status_callback: Option<K>,
 	handle_new_peer_callback: H,
 ) -> Result<(), Error>
 where
 	F: Fn(Option<String>),
 	G: Fn(&Error) -> bool, // return true if want to exit on falure
 	H: Fn(TcpDataStream, Option<PeerAddr>),
+	K: Fn(bool) + Send + 'static + std::marker::Sync, // return true if want to exit on falure
 {
 	//
 	//let handle_new_peer_callback = Arc::new(handle_new_peer_callback);
 	let arti_streams = AtomicI64::new(0);
 
+	let service_status_callback = Arc::new(service_status_callback);
+	if let Some(f) = &(*service_status_callback) {
+		f(false);
+	};
+
 	loop {
 		if stop_state.is_stopped() {
 			break;
@@ -119,6 +133,7 @@ where
 				let stop_state2 = stop_state.clone();
 				let context_id2 = context_id;
 				let service_name2 = String::from(service_name);
+				let service_status_callback2 = service_status_callback.clone();
 
 				let monitoring = thread::Builder::new()
 					.name(format!(
@@ -127,8 +142,13 @@ where
 					))
 					.spawn(move || {
 						let mut last_running_time = Instant::now();
+						if let Some(f) = &(*service_status_callback2) {
+							f(false);
+						};
 						loop {
 							if stop_state2.is_stopped() {
+								arti::unregister_arti_active_object(&onion_service_object);
+								drop(onion_service);
 								break;
 							}
 							let need_arti_restart = {
@@ -152,10 +172,16 @@ where
 										| State::DegradedUnreachable
 										| State::Running => {
 											last_running_time = Instant::now();
+											if let Some(f) = &(*service_status_callback2) {
+												f(true);
+											};
 											false
 										}
 										State::Broken => true,
 										_ => {
+											if let Some(f) = &(*service_status_callback2) {
+												f(false);
+											};
 											let elapsed =
 												Instant::now().duration_since(last_running_time);
 											// Giving 3 minutes to arti to restore
@@ -172,8 +198,8 @@ where
 
 							if need_arti_restart || arti::is_arti_restarting() {
 								drop(onion_service);
-								arti::request_arti_restart();
 								arti::unregister_arti_active_object(&onion_service_object);
+								arti::request_arti_restart();
 								break;
 							}
 							for _ in 0..30 {
@@ -183,6 +209,9 @@ where
 								thread::sleep(Duration::from_secs(1));
 							}
 						}
+						if let Some(f) = &(*service_status_callback2) {
+							f(false);
+						};
 					})
 					.expect(&format!(
 						"Unable to start {} onion_service_checker thread",