Skip to content

Commit

Permalink
dev: cleanup service bootstraping
Browse files Browse the repository at this point in the history
  • Loading branch information
da2ce7 committed Dec 31, 2023
1 parent 933eacb commit b53a219
Show file tree
Hide file tree
Showing 29 changed files with 806 additions and 849 deletions.
20 changes: 10 additions & 10 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

3 changes: 2 additions & 1 deletion cSpell.json
Original file line number Diff line number Diff line change
Expand Up @@ -129,7 +129,8 @@
"Xtorrent",
"Xunlei",
"xxxxxxxxxxxxxxxxxxxxd",
"yyyyyyyyyyyyyyyyyyyyd"
"yyyyyyyyyyyyyyyyyyyyd",
"nvCFlJCq7fz7Qx6KoKTDiMZvns8l5Kw7"
],
"enableFiletypes": [
"dockerfile",
Expand Down
4 changes: 2 additions & 2 deletions packages/configuration/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -239,7 +239,7 @@ use config::{Config, ConfigError, File, FileFormat};
use serde::{Deserialize, Serialize};
use serde_with::{serde_as, NoneAsEmptyString};
use thiserror::Error;
use torrust_tracker_located_error::{Located, LocatedError};
use torrust_tracker_located_error::{DynError, Located, LocatedError};
use torrust_tracker_primitives::{DatabaseDriver, TrackerMode};

/// Information required for loading config
Expand Down Expand Up @@ -289,7 +289,7 @@ impl Info {

fs::read_to_string(config_path)
.map_err(|e| Error::UnableToLoadFromConfigFile {
source: (Arc::new(e) as Arc<dyn std::error::Error + Send + Sync>).into(),
source: (Arc::new(e) as DynError).into(),

Check warning on line 292 in packages/configuration/src/lib.rs

View check run for this annotation

Codecov / codecov/patch

packages/configuration/src/lib.rs#L292

Added line #L292 was not covered by tests
})?
.parse()
.map_err(|_e: std::convert::Infallible| Error::Infallible)?
Expand Down
8 changes: 6 additions & 2 deletions packages/located-error/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,10 @@ use std::error::Error;
use std::panic::Location;
use std::sync::Arc;

use log::debug;

pub type DynError = Arc<dyn std::error::Error + Send + Sync>;

/// A generic wrapper around an error.
///
/// Where `E` is the inner error (source error).
Expand Down Expand Up @@ -90,13 +94,13 @@ where
source: Arc::new(self.0),
location: Box::new(*std::panic::Location::caller()),
};
log::debug!("{e}");
debug!("{e}");
e
}
}

#[allow(clippy::from_over_into)]
impl<'a> Into<LocatedError<'a, dyn std::error::Error + Send + Sync>> for Arc<dyn std::error::Error + Send + Sync> {
impl<'a> Into<LocatedError<'a, dyn std::error::Error + Send + Sync>> for DynError {
#[track_caller]
fn into(self) -> LocatedError<'a, dyn std::error::Error + Send + Sync> {
LocatedError {
Expand Down
16 changes: 8 additions & 8 deletions src/app.rs
Original file line number Diff line number Diff line change
Expand Up @@ -28,8 +28,7 @@ use tokio::task::JoinHandle;
use torrust_tracker_configuration::Configuration;

use crate::bootstrap::jobs::{health_check_api, http_tracker, torrent_cleanup, tracker_apis, udp_tracker};
use crate::core;
use crate::servers::http::Version;
use crate::{core, servers};

/// # Panics
///
Expand Down Expand Up @@ -68,21 +67,22 @@ pub async fn start(config: Arc<Configuration>, tracker: Arc<core::Tracker>) -> V
udp_tracker_config.bind_address, config.mode
);
} else {
jobs.push(udp_tracker::start_job(udp_tracker_config, tracker.clone()));
jobs.push(udp_tracker::start_job(udp_tracker_config, tracker.clone()).await);

Check warning on line 70 in src/app.rs

View check run for this annotation

Codecov / codecov/patch

src/app.rs#L70

Added line #L70 was not covered by tests
}
}

// Start the HTTP blocks
for http_tracker_config in &config.http_trackers {
if !http_tracker_config.enabled {
continue;
}
jobs.push(http_tracker::start_job(http_tracker_config, tracker.clone(), Version::V1).await);
if let Some(job) = http_tracker::start_job(http_tracker_config, tracker.clone(), servers::http::Version::V1).await {
jobs.push(job);
};

Check warning on line 78 in src/app.rs

View check run for this annotation

Codecov / codecov/patch

src/app.rs#L76-L78

Added lines #L76 - L78 were not covered by tests
}

// Start HTTP API
if config.http_api.enabled {
jobs.push(tracker_apis::start_job(&config.http_api, tracker.clone()).await);
if let Some(job) = tracker_apis::start_job(&config.http_api, tracker.clone(), servers::apis::Version::V1).await {
jobs.push(job);
};

Check warning on line 85 in src/app.rs

View check run for this annotation

Codecov / codecov/patch

src/app.rs#L83-L85

Added lines #L83 - L85 were not covered by tests
}

// Start runners to remove torrents without peers, every interval
Expand Down
25 changes: 6 additions & 19 deletions src/bootstrap/jobs/health_check_api.rs
Original file line number Diff line number Diff line change
Expand Up @@ -6,36 +6,23 @@
//! The [`health_check_api::start_job`](crate::bootstrap::jobs::health_check_api::start_job)
//! function spawns a new asynchronous task, that tasks is the "**launcher**".
//! The "**launcher**" starts the actual server and sends a message back
//! to the main application. The main application waits until receives
//! the message [`ApiServerJobStarted`]
//! from the "**launcher**".
//! to the main application.
//!
//! The "**launcher**" is an intermediary thread that decouples the Health Check
//! API server from the process that handles it.
//!
//! Refer to the [configuration documentation](https://docs.rs/torrust-tracker-configuration)
//! for the API configuration options.
use std::net::SocketAddr;
use std::sync::Arc;

use log::info;
use tokio::sync::oneshot;
use tokio::task::JoinHandle;
use torrust_tracker_configuration::Configuration;

use super::Started;
use crate::servers::health_check_api::server;

/// This is the message that the "launcher" spawned task sends to the main
/// application process to notify the API server was successfully started.
///
/// > **NOTICE**: it does not mean the API server is ready to receive requests.
/// It only means the new server started. It might take some time to the server
/// to be ready to accept request.
#[derive(Debug)]
pub struct ApiServerJobStarted {
pub bound_addr: SocketAddr,
}

/// This function starts a new Health Check API server with the provided
/// configuration.
///
Expand All @@ -53,22 +40,22 @@ pub async fn start_job(config: Arc<Configuration>) -> JoinHandle<()> {
.parse::<std::net::SocketAddr>()
.expect("Health Check API bind_address invalid.");

let (tx, rx) = oneshot::channel::<ApiServerJobStarted>();
let (tx_start, rx_start) = oneshot::channel::<Started>();

Check warning on line 43 in src/bootstrap/jobs/health_check_api.rs

View check run for this annotation

Codecov / codecov/patch

src/bootstrap/jobs/health_check_api.rs#L43

Added line #L43 was not covered by tests

// Run the API server
let join_handle = tokio::spawn(async move {
info!("Starting Health Check API server: http://{}", bind_addr);

let handle = server::start(bind_addr, tx, config.clone());
let handle = server::start(bind_addr, tx_start, config.clone());

Check warning on line 49 in src/bootstrap/jobs/health_check_api.rs

View check run for this annotation

Codecov / codecov/patch

src/bootstrap/jobs/health_check_api.rs#L49

Added line #L49 was not covered by tests

if let Ok(()) = handle.await {
info!("Health Check API server on http://{} stopped", bind_addr);
}
});

// Wait until the API server job is running
match rx.await {
Ok(_msg) => info!("Torrust Health Check API server started"),
match rx_start.await {
Ok(msg) => info!("Torrust Health Check API server started on socket: {}", msg.address),

Check warning on line 58 in src/bootstrap/jobs/health_check_api.rs

View check run for this annotation

Codecov / codecov/patch

src/bootstrap/jobs/health_check_api.rs#L57-L58

Added lines #L57 - L58 were not covered by tests
Err(e) => panic!("the Health Check API server was dropped: {e}"),
}

Expand Down
103 changes: 47 additions & 56 deletions src/bootstrap/jobs/http_tracker.rs
Original file line number Diff line number Diff line change
Expand Up @@ -7,88 +7,79 @@
//!
//! The [`http_tracker::start_job`](crate::bootstrap::jobs::http_tracker::start_job) function spawns a new asynchronous task,
//! that tasks is the "**launcher**". The "**launcher**" starts the actual server and sends a message back to the main application.
//! The main application waits until receives the message [`ServerJobStarted`] from the "**launcher**".
//!
//! The "**launcher**" is an intermediary thread that decouples the HTTP servers from the process that handles it. The HTTP could be used independently in the future.
//! In that case it would not need to notify a parent process.
use std::net::SocketAddr;
use std::sync::Arc;

use axum_server::tls_rustls::RustlsConfig;
use log::info;
use tokio::sync::oneshot;
use tokio::task::JoinHandle;
use torrust_tracker_configuration::HttpTracker;

use super::make_rust_tls;
use crate::core;
use crate::servers::http::v1::launcher;
use crate::servers::http::server::{HttpServer, Launcher};
use crate::servers::http::Version;

/// This is the message that the "**launcher**" spawned task sends to the main application process to notify that the HTTP server was successfully started.
///
/// > **NOTICE**: it does not mean the HTTP server is ready to receive requests. It only means the new server started. It might take some time to the server to be ready to accept request.
#[derive(Debug)]
pub struct ServerJobStarted();

/// It starts a new HTTP server with the provided configuration and version.
///
/// Right now there is only one version but in the future we could support more than one HTTP tracker version at the same time.
/// This feature allows supporting breaking changes on `BitTorrent` BEPs.
pub async fn start_job(config: &HttpTracker, tracker: Arc<core::Tracker>, version: Version) -> JoinHandle<()> {
match version {
Version::V1 => start_v1(config, tracker.clone()).await,
}
}

///
/// # Panics
///
/// It would panic if the `config::HttpTracker` struct would contain inappropriate values.
async fn start_v1(config: &HttpTracker, tracker: Arc<core::Tracker>) -> JoinHandle<()> {
let bind_addr = config
.bind_address
.parse::<std::net::SocketAddr>()
.expect("Tracker API bind_address invalid.");
let ssl_enabled = config.ssl_enabled;
let ssl_cert_path = config.ssl_cert_path.clone();
let ssl_key_path = config.ssl_key_path.clone();

let (tx, rx) = oneshot::channel::<ServerJobStarted>();

// Run the API server
let join_handle = tokio::spawn(async move {
if !ssl_enabled {
info!("Starting Torrust HTTP tracker server on: http://{}", bind_addr);

let handle = launcher::start(bind_addr, tracker);
///
pub async fn start_job(config: &HttpTracker, tracker: Arc<core::Tracker>, version: Version) -> Option<JoinHandle<()>> {
if config.enabled {
let socket = config
.bind_address
.parse::<std::net::SocketAddr>()
.expect("Tracker API bind_address invalid.");

let tls = make_rust_tls(config.ssl_enabled, &config.ssl_cert_path, &config.ssl_key_path)
.await
.map(|tls| tls.expect("tls config failed"));

match version {
Version::V1 => Some(start_v1(socket, tls, tracker.clone()).await),
}
} else {
info!("Note: Not loading Http Tracker Service, Not Enabled in Configuration.");
None

Check warning on line 51 in src/bootstrap/jobs/http_tracker.rs

View check run for this annotation

Codecov / codecov/patch

src/bootstrap/jobs/http_tracker.rs#L50-L51

Added lines #L50 - L51 were not covered by tests
}
}

tx.send(ServerJobStarted())
.expect("the HTTP tracker server should not be dropped");
async fn start_v1(socket: SocketAddr, tls: Option<RustlsConfig>, tracker: Arc<core::Tracker>) -> JoinHandle<()> {
let server = HttpServer::new(Launcher::new(socket, tls))
.start(tracker)
.await
.expect("Failed to start Server");

if let Ok(()) = handle.await {
info!("Torrust HTTP tracker server on http://{} stopped", bind_addr);
}
} else if ssl_enabled && ssl_cert_path.is_some() && ssl_key_path.is_some() {
info!("Starting Torrust HTTP tracker server on: https://{}", bind_addr);
tokio::spawn(async move {
server.state.task.await.expect("failed to finish service");

Check warning on line 62 in src/bootstrap/jobs/http_tracker.rs

View check run for this annotation

Codecov / codecov/patch

src/bootstrap/jobs/http_tracker.rs#L62

Added line #L62 was not covered by tests
})
}

let ssl_config = RustlsConfig::from_pem_file(ssl_cert_path.unwrap(), ssl_key_path.unwrap())
.await
.unwrap();
#[cfg(test)]
mod tests {
use std::sync::Arc;

let handle = launcher::start_tls(bind_addr, ssl_config, tracker);
use torrust_tracker_test_helpers::configuration::ephemeral_mode_public;

tx.send(ServerJobStarted())
.expect("the HTTP tracker server should not be dropped");
use crate::bootstrap::app::initialize_with_configuration;
use crate::bootstrap::jobs::http_tracker::start_job;
use crate::servers::http::Version;

if let Ok(()) = handle.await {
info!("Torrust HTTP tracker server on https://{} stopped", bind_addr);
}
}
});
#[tokio::test]
async fn it_should_start_http_tracker() {
let cfg = Arc::new(ephemeral_mode_public());
let config = &cfg.http_trackers[0];
let tracker = initialize_with_configuration(&cfg);
let version = Version::V1;

// Wait until the HTTP tracker server job is running
match rx.await {
Ok(_msg) => info!("Torrust HTTP tracker server started"),
Err(e) => panic!("the HTTP tracker server was dropped: {e}"),
start_job(config, tracker, version).await.expect("some https server");
}

join_handle
}
Loading

0 comments on commit b53a219

Please sign in to comment.