Merge remote-tracking branch 'grin/staging' into grim

Author: ardocrat

Cargo.lock

@@ -143,23 +143,37 @@ impl ApiServer {
 		conf: Option<TLSConfig>,
 		api_chan: (mpsc::Sender<()>, mpsc::Receiver<()>),
 	) -> Result<thread::JoinHandle<()>, Error> {
-		let _ = rustls::crypto::ring::default_provider().install_default();
-
 		if self.shutdown_sender.is_some() {
 			return Err(Error::Internal(
 				"Can't start API server, it's running already".to_string(),
 			));
 		}
-		self.shutdown_sender = Some(api_chan.0);
+
+		let _ = rustls::crypto::ring::default_provider().install_default();
+
+		// Check if provided address is free.
+		let listener = match std::net::TcpListener::bind(addr) {
+			Ok(l) => {
+				l.set_nonblocking(true)
+					.map_err(|e| Error::Internal(format!("API listener binding error: {}", e)))?;
+				l
+			}
+			Err(e) => {
+				error!("API listener binding error: {}", e);
+				return Err(Error::Internal(e.to_string()));
+			}
+		};
 
 		let tls = match conf {
 			Some(conf) => Some(TlsAcceptor::from(conf.build_server_config()?)),
 			None => None,
 		};
-		thread::Builder::new()
+		let res = thread::Builder::new()
 			.name("apis".to_string())
-			.spawn(move || start_server(addr, router, api_chan.1, tls))
-			.map_err(|_| Error::Internal("failed to spawn API thread".to_string()))
+			.spawn(move || start_server(listener, router, api_chan.1, tls))
+			.map_err(|_| Error::Internal("failed to spawn API thread".to_string()))?;
+		self.shutdown_sender = Some(api_chan.0);
+		Ok(res)
 	}
 
 	/// Stops the API server.
@@ -184,7 +198,7 @@ impl ApiServer {
 
 /// Start API server with optional TLS support.
 fn start_server(
-	addr: SocketAddr,
+	l: std::net::TcpListener,
 	router: Router,
 	rx: mpsc::Receiver<()>,
 	tls: Option<TlsAcceptor>,
@@ -194,77 +208,77 @@ fn start_server(
 		// When this signal completes, start shutdown.
 		let mut signal = std::pin::pin!(shutdown_signal(rx));
 
-		// Start server loop.
-		match TcpListener::bind(addr).await {
-			Ok(l) => {
-				loop {
-					tokio::select! {
-						Ok(s) = async {
-							match l.accept().await {
-								Ok((s, _)) => Ok::<Option<tokio::net::TcpStream>, Error>(Some(s)),
-								Err(e) => {
-									error!("Failed to accept connection: {e:#}");
-									Ok(None)
-								}
-							}
-						} => {
-							if let Some(s) = s {
-								if let Some(tls) = tls.clone() {
-									let router = router.clone();
-									let watcher = graceful.watcher();
-									tokio::spawn(async move {
-										let handshake = timeout(TLS_HANDSHAKE_TIMEOUT, tls.accept(s));
-										let tls_stream = match handshake.await {
-											Ok(Ok(tls_stream)) => tls_stream,
-											Ok(Err(err)) => {
-												error!("failed to perform TLS handshake: {err:#}");
-												return;
-											}
-											Err(_) => {
-												error!("TLS handshake timed out");
-												return;
-											}
-										};
-										let io = TokioIo::new(tls_stream);
-										let conn = http1::Builder::new().serve_connection(io, router);
-										if let Err(e) = watcher.watch(conn).await {
-											error!("API TLS server error: {:?}", e);
-										}
-									});
-								} else {
-									let io = TokioIo::new(s);
-									let conn = http1::Builder::new().serve_connection(io, router.clone());
-									let fut = graceful.watch(conn);
-									tokio::spawn(async move {
-										if let Err(e) = fut.await {
-											error!("API HTTP server error: {:?}", e);
-										}
-									});
-								};
-							} else {
-								continue;
-							}
-						}
-						_ = &mut signal => {
-							drop(l);
-							break;
+		let l = match TcpListener::from_std(l) {
+			Ok(l) => l,
+			Err(e) => {
+				error!("HTTP API server error: {}", e);
+				return;
+			}
+		};
+
+		loop {
+			tokio::select! {
+				Ok(s) = async {
+					match l.accept().await {
+						Ok((s, _)) => Ok::<Option<tokio::net::TcpStream>, Error>(Some(s)),
+						Err(e) => {
+							error!("Failed to accept connection: {e:#}");
+							Ok(None)
 						}
 					}
-				}
-
-				// Now start the shutdown and wait for them to complete
-				// Also start a timeout to limit how long to wait.
-				tokio::select! {
-					_ = graceful.shutdown() => {
-						warn!("API server gracefully stopped");
-					},
-					_ = sleep(GRACEFUL_SHUTDOWN_TIMEOUT) => {
-						warn!("API server timed out wait for all connections to close");
+				} => {
+					if let Some(s) = s {
+						if let Some(tls) = tls.clone() {
+							let router = router.clone();
+							let watcher = graceful.watcher();
+							tokio::spawn(async move {
+								let handshake = timeout(TLS_HANDSHAKE_TIMEOUT, tls.accept(s));
+								let tls_stream = match handshake.await {
+									Ok(Ok(tls_stream)) => tls_stream,
+									Ok(Err(err)) => {
+										error!("failed to perform TLS handshake: {err:#}");
+										return;
+									}
+									Err(_) => {
+										error!("TLS handshake timed out");
+										return;
+									}
+								};
+								let io = TokioIo::new(tls_stream);
+								let conn = http1::Builder::new().serve_connection(io, router);
+								if let Err(e) = watcher.watch(conn).await {
+									error!("API TLS server error: {:?}", e);
+								}
+							});
+						} else {
+							let io = TokioIo::new(s);
+							let conn = http1::Builder::new().serve_connection(io, router.clone());
+							let fut = graceful.watch(conn);
+							tokio::spawn(async move {
+								if let Err(e) = fut.await {
+									error!("API HTTP server error: {:?}", e);
+								}
+							});
+						};
+					} else {
+						continue;
 					}
 				}
+				_ = &mut signal => {
+					drop(l);
+					break;
+				}
 			}
-			Err(e) => {
-				error!("API listener binding error: {}", e);
+		}
+
+		// Now start the shutdown and wait for them to complete
+		// Also start a timeout to limit how long to wait.
+		tokio::select! {
+			_ = graceful.shutdown() => {
+				warn!("API server gracefully stopped");
+			},
+			_ = sleep(GRACEFUL_SHUTDOWN_TIMEOUT) => {
+				warn!("API server timed out wait for all connections to close");
 			}
 		}
 	};

api/src/rest.rs

@@ -107,6 +107,24 @@ fn test_start_api() {
 	thread::sleep(time::Duration::from_millis(1_000));
 }
 
+#[test]
+fn test_start_api_address_in_use() {
+	let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+	let addr = listener.local_addr().unwrap();
+	let mut server = ApiServer::new();
+
+	assert!(server
+		.start(addr, build_router(), None, mpsc::channel::<()>(1))
+		.is_err());
+
+	drop(listener);
+	let handle = server
+		.start(addr, build_router(), None, mpsc::channel::<()>(1))
+		.unwrap();
+	assert!(server.stop());
+	handle.join().unwrap();
+}
+
 // To enable this test you need a trusted PKCS12 (p12) certificate bundle
 // Hyper-tls client doesn't accept self-signed certificates. The easiest way is to use mkcert
 // https://github.com/FiloSottile/mkcert to install CA and generate a certificate on your local machine.

api/tests/rest.rs

@@ -434,8 +434,8 @@ impl Peer {
 	}
 
 	/// Stops the peer
-	pub fn stop(&self) {
-		debug!("Stopping peer {:?}", self.info.addr);
+	pub fn stop(&self, reason: &str) {
+		debug!("Stopping peer {:?}, reason: {}", self.info.addr, reason);
 		match self.stop_handle.try_lock() {
 			Some(handle) => handle.stop(),
 			None => error!("can't get stop lock for peer"),