refactor: [#1358] inject event sender in Swarm type

It required to use `tokio::sync::Mutex` for the `SwarmHandle` (`Arc<Mutex<Swarm>>`). Otherwise it's not safe to pass the Swarm lock between threads.

Author: josecelano

Cargo.lock

@@ -787,7 +787,8 @@ mod for_all_config_modes {
                 .container
                 .tracker_core_container
                 .in_memory_torrent_repository
-                .get_torrent_peers(&info_hash);
+                .get_torrent_peers(&info_hash)
+                .await;
             let peer_addr = peers[0].peer_addr;
 
             assert_eq!(peer_addr.ip(), client_ip);
@@ -829,7 +830,8 @@ mod for_all_config_modes {
                 .container
                 .tracker_core_container
                 .in_memory_torrent_repository
-                .get_torrent_peers(&info_hash);
+                .get_torrent_peers(&info_hash)
+                .await;
             let peer_addr = peers[0].peer_addr;
 
             assert_eq!(
@@ -878,7 +880,8 @@ mod for_all_config_modes {
                 .container
                 .tracker_core_container
                 .in_memory_torrent_repository
-                .get_torrent_peers(&info_hash);
+                .get_torrent_peers(&info_hash)
+                .await;
             let peer_addr = peers[0].peer_addr;
 
             assert_eq!(
@@ -925,7 +928,8 @@ mod for_all_config_modes {
                 .container
                 .tracker_core_container
                 .in_memory_torrent_repository
-                .get_torrent_peers(&info_hash);
+                .get_torrent_peers(&info_hash)
+                .await;
             let peer_addr = peers[0].peer_addr;
 
             assert_eq!(peer_addr.ip(), IpAddr::from_str("150.172.238.178").unwrap());

packages/axum-http-tracker-server/tests/server/v1/contract.rs

@@ -33,7 +33,7 @@ pub async fn get_torrent_handler(
 ) -> Response {
     match InfoHash::from_str(&info_hash.0) {
         Err(_) => invalid_info_hash_param_response(&info_hash.0),
-        Ok(info_hash) => match get_torrent_info(&in_memory_torrent_repository, &info_hash) {
+        Ok(info_hash) => match get_torrent_info(&in_memory_torrent_repository, &info_hash).await {
             Some(info) => torrent_info_response(info).into_response(),
             None => torrent_not_known_response(),
         },
@@ -85,14 +85,19 @@ pub async fn get_torrents_handler(
     tracing::debug!("pagination: {:?}", pagination);
 
     if pagination.0.info_hashes.is_empty() {
-        torrent_list_response(&get_torrents_page(
-            &in_memory_torrent_repository,
-            Some(&Pagination::new_with_options(pagination.0.offset, pagination.0.limit)),
-        ))
+        torrent_list_response(
+            &get_torrents_page(
+                &in_memory_torrent_repository,
+                Some(&Pagination::new_with_options(pagination.0.offset, pagination.0.limit)),
+            )
+            .await,
+        )
         .into_response()
     } else {
         match parse_info_hashes(pagination.0.info_hashes) {
-            Ok(info_hashes) => torrent_list_response(&get_torrents(&in_memory_torrent_repository, &info_hashes)).into_response(),
+            Ok(info_hashes) => {
+                torrent_list_response(&get_torrents(&in_memory_torrent_repository, &info_hashes).await).into_response()
+            }
             Err(err) => match err {
                 QueryParamError::InvalidInfoHash { info_hash } => invalid_info_hash_param_response(&info_hash),
             },

packages/axum-rest-tracker-api-server/src/v1/context/torrent/handlers.rs

@@ -47,7 +47,7 @@ pub async fn get_metrics(
     in_memory_torrent_repository: Arc<InMemoryTorrentRepository>,
     stats_repository: Arc<Repository>,
 ) -> TrackerMetrics {
-    let torrents_metrics = in_memory_torrent_repository.get_aggregate_swarm_metadata();
+    let torrents_metrics = in_memory_torrent_repository.get_aggregate_swarm_metadata().await;
     let stats = stats_repository.get_stats().await;
 
     TrackerMetrics {

packages/http-tracker-core/src/statistics/services.rs

@@ -32,7 +32,7 @@ pub async fn get_metrics(
     http_stats_repository: Arc<bittorrent_http_tracker_core::statistics::repository::Repository>,
     udp_server_stats_repository: Arc<udp_server_statistics::repository::Repository>,
 ) -> TrackerMetrics {
-    let torrents_metrics = in_memory_torrent_repository.get_aggregate_swarm_metadata();
+    let torrents_metrics = in_memory_torrent_repository.get_aggregate_swarm_metadata().await;
     let udp_banned_ips_total = ban_service.read().await.get_banned_ips_total();
     let http_stats = http_stats_repository.get_stats().await;
     let udp_server_stats = udp_server_stats_repository.get_stats().await;

packages/rest-tracker-api-core/src/statistics/services.rs

@@ -19,6 +19,7 @@ version.workspace = true
 aquatic_udp_protocol = "0"
 bittorrent-primitives = "0.1.0"
 crossbeam-skiplist = "0"
+futures = "0"
 serde = "1.0.219"
 thiserror = "2.0.12"
 tokio = { version = "1", features = ["macros", "net", "rt-multi-thread", "signal", "sync"] }

packages/torrent-repository/Cargo.toml

@@ -4,8 +4,9 @@ pub mod statistics;
 pub mod swarm;
 pub mod swarms;
 
-use std::sync::{Arc, Mutex, MutexGuard};
+use std::sync::Arc;
 
+use tokio::sync::Mutex;
 use torrust_tracker_clock::clock;
 
 pub type Swarms = swarms::Swarms;
@@ -24,16 +25,6 @@ pub(crate) type CurrentClock = clock::Stopped;
 
 pub const TORRENT_REPOSITORY_LOG_TARGET: &str = "TORRENT_REPOSITORY";
 
-pub trait LockTrackedTorrent {
-    fn lock_or_panic(&self) -> MutexGuard<'_, Swarm>;
-}
-
-impl LockTrackedTorrent for SwarmHandle {
-    fn lock_or_panic(&self) -> MutexGuard<'_, Swarm> {
-        self.lock().expect("can't acquire lock for tracked torrent handle")
-    }
-}
-
 #[cfg(test)]
 pub(crate) mod tests {
     use std::net::{IpAddr, Ipv4Addr, SocketAddr};