diff --git a/cSpell.json b/cSpell.json index d09db93b7..9602ba39b 100644 --- a/cSpell.json +++ b/cSpell.json @@ -32,6 +32,7 @@ "Containerfile", "curr", "Cyberneering", + "datagram", "datetime", "Dijke", "distroless", @@ -79,6 +80,7 @@ "nonroot", "Norberg", "numwant", + "nvCFlJCq7fz7Qx6KoKTDiMZvns8l5Kw7", "oneshot", "ostr", "Pando", @@ -129,8 +131,7 @@ "Xtorrent", "Xunlei", "xxxxxxxxxxxxxxxxxxxxd", - "yyyyyyyyyyyyyyyyyyyyd", - "nvCFlJCq7fz7Qx6KoKTDiMZvns8l5Kw7" + "yyyyyyyyyyyyyyyyyyyyd" ], "enableFiletypes": [ "dockerfile", diff --git a/src/bootstrap/jobs/health_check_api.rs b/src/bootstrap/jobs/health_check_api.rs index a49f612e8..9fed56435 100644 --- a/src/bootstrap/jobs/health_check_api.rs +++ b/src/bootstrap/jobs/health_check_api.rs @@ -44,18 +44,18 @@ pub async fn start_job(config: Arc) -> JoinHandle<()> { // Run the API server let join_handle = tokio::spawn(async move { - info!("Starting Health Check API server: http://{}", bind_addr); + info!(target: "Health Check API", "Starting on: http://{}", bind_addr); let handle = server::start(bind_addr, tx_start, config.clone()); if let Ok(()) = handle.await { - info!("Health Check API server on http://{} stopped", bind_addr); + info!(target: "Health Check API", "Stopped server running on: http://{}", bind_addr); } }); // Wait until the API server job is running match rx_start.await { - Ok(msg) => info!("Torrust Health Check API server started on: http://{}", msg.address), + Ok(msg) => info!(target: "Health Check API", "Started on: http://{}", msg.address), Err(e) => panic!("the Health Check API server was dropped: {e}"), } diff --git a/src/bootstrap/jobs/udp_tracker.rs b/src/bootstrap/jobs/udp_tracker.rs index 5911bdf95..20ef0c793 100644 --- a/src/bootstrap/jobs/udp_tracker.rs +++ b/src/bootstrap/jobs/udp_tracker.rs @@ -8,6 +8,7 @@ //! for the configuration options. use std::sync::Arc; +use log::debug; use tokio::task::JoinHandle; use torrust_tracker_configuration::UdpTracker; @@ -36,10 +37,20 @@ pub async fn start_job(config: &UdpTracker, tracker: Arc) -> Join .expect("it should be able to start the udp tracker"); tokio::spawn(async move { + debug!(target: "UDP Tracker", "Wait for launcher (UDP service) to finish ..."); + debug!(target: "UDP Tracker", "Is halt channel closed before waiting?: {}", server.state.halt_task.is_closed()); + + assert!( + !server.state.halt_task.is_closed(), + "Halt channel for UDP tracker should be open" + ); + server .state .task .await .expect("it should be able to join to the udp tracker task"); + + debug!(target: "UDP Tracker", "Is halt channel closed after finishing the server?: {}", server.state.halt_task.is_closed()); }) } diff --git a/src/servers/apis/server.rs b/src/servers/apis/server.rs index 5df1d76fd..f4fdf8994 100644 --- a/src/servers/apis/server.rs +++ b/src/servers/apis/server.rs @@ -175,6 +175,8 @@ impl Launcher { let tls = self.tls.clone(); let protocol = if tls.is_some() { "https" } else { "http" }; + info!(target: "API", "Starting on {protocol}://{}", address); + let running = Box::pin(async { match tls { Some(tls) => axum_server::from_tcp_rustls(socket, tls) @@ -190,7 +192,7 @@ impl Launcher { } }); - info!(target: "API", "API server started on {protocol}://{}", address); + info!(target: "API", "Started on {protocol}://{}", address); tx_start .send(Started { address }) diff --git a/src/servers/http/server.rs b/src/servers/http/server.rs index 904ccdcf5..0a4b687b5 100644 --- a/src/servers/http/server.rs +++ b/src/servers/http/server.rs @@ -48,7 +48,7 @@ impl Launcher { tokio::task::spawn(graceful_shutdown( handle.clone(), rx_halt, - format!("Shutting down http server on socket address: {address}"), + format!("Shutting down HTTP server on socket address: {address}"), )); let tls = self.tls.clone(); diff --git a/src/servers/udp/server.rs b/src/servers/udp/server.rs index a0af55101..22cdf6357 100644 --- a/src/servers/udp/server.rs +++ b/src/servers/udp/server.rs @@ -120,23 +120,30 @@ impl UdpServer { let (tx_start, rx_start) = tokio::sync::oneshot::channel::(); let (tx_halt, rx_halt) = tokio::sync::oneshot::channel::(); + assert!(!tx_halt.is_closed(), "Halt channel for UDP tracker should be open"); + let launcher = self.state.launcher; let task = tokio::spawn(async move { - launcher.start(tracker, tx_start, rx_halt).await; + debug!(target: "UDP Tracker", "Launcher starting ..."); + + let starting = launcher.start(tracker, tx_start, rx_halt).await; + + starting.await.expect("UDP server should have started running"); + launcher }); + let binding = rx_start.await.expect("unable to start service").address; + let running_udp_server: UdpServer = UdpServer { state: Running { - binding: rx_start.await.expect("unable to start service").address, + binding, halt_task: tx_halt, task, }, }; - info!("Running UDP Tracker on Socket: {}", running_udp_server.state.binding); - Ok(running_udp_server) } } @@ -202,41 +209,62 @@ impl Udp { tx_start: Sender, rx_halt: Receiver, ) -> JoinHandle<()> { - let binding = Arc::new(UdpSocket::bind(bind_to).await.expect("Could not bind to {self.socket}.")); - let address = binding.local_addr().expect("Could not get local_addr from {binding}."); + let socket = Arc::new(UdpSocket::bind(bind_to).await.expect("Could not bind to {self.socket}.")); + let address = socket.local_addr().expect("Could not get local_addr from {binding}."); + + info!(target: "UDP Tracker", "Starting on: udp://{}", address); let running = tokio::task::spawn(async move { - let halt = async move { - shutdown_signal_with_message(rx_halt, format!("Halting Http Service Bound to Socket: {address}")).await; + let halt = tokio::task::spawn(async move { + debug!(target: "UDP Tracker", "Waiting for halt signal for socket address: udp://{address} ..."); + + shutdown_signal_with_message( + rx_halt, + format!("Shutting down UDP server on socket address: udp://{address}"), + ) + .await; + }); + + let listen = async move { + debug!(target: "UDP Tracker", "Waiting for packets on socket address: udp://{address} ..."); + + loop { + let mut data = [0; MAX_PACKET_SIZE]; + let socket_clone = socket.clone(); + + match socket_clone.recv_from(&mut data).await { + Ok((valid_bytes, remote_addr)) => { + let payload = data[..valid_bytes].to_vec(); + + debug!(target: "UDP Tracker", "Received {} bytes", payload.len()); + debug!(target: "UDP Tracker", "From: {}", &remote_addr); + debug!(target: "UDP Tracker", "Payload: {:?}", payload); + + let response = handle_packet(remote_addr, payload, &tracker).await; + + Udp::send_response(socket_clone, remote_addr, response).await; + } + Err(err) => { + error!("Error reading UDP datagram from socket. Error: {:?}", err); + } + } + } }; pin_mut!(halt); + pin_mut!(listen); - loop { - let mut data = [0; MAX_PACKET_SIZE]; - let binding = binding.clone(); - - tokio::select! { - () = & mut halt => {}, - - Ok((valid_bytes, remote_addr)) = binding.recv_from(&mut data) => { - let payload = data[..valid_bytes].to_vec(); - - debug!("Received {} bytes", payload.len()); - debug!("From: {}", &remote_addr); - debug!("Payload: {:?}", payload); + tx_start + .send(Started { address }) + .expect("the UDP Tracker service should not be dropped"); - let response = handle_packet(remote_addr, payload, &tracker).await; - - Udp::send_response(binding, remote_addr, response).await; - } - } + tokio::select! { + _ = & mut halt => { debug!(target: "UDP Tracker", "Halt signal spawned task stopped on address: udp://{address}"); }, + () = & mut listen => { debug!(target: "UDP Tracker", "Socket listener stopped on address: udp://{address}"); }, } }); - tx_start - .send(Started { address }) - .expect("the UDP Tracker service should not be dropped"); + info!(target: "UDP Tracker", "Started on: udp://{}", address); running }