Merge torrust#1356: Fix bug: number of downloads for a torrent is reset

06ef1d5 fix: [torrust#1264] number of downloads preset when torrent is persisted (Jose Celano)
8f67f12 fix: [torrust#1264] partially. The correct number of downloads is persited (Jose Celano)
6beec3a refactor: [torrust#1264] add a database method to increase the number of downloads for a torrent (Jose Celano)
8a4dba3 refactor: [torrust#1264] rename variables (Jose Celano)
2fb1c6f reafactor: [torrust#1264] add a DB methog to get the number of completed downloads for one torrent (Jose Celano)
c5db71a refactor: [torrust#1264] remove unnecessary fn call (Jose Celano)
820329b refactor: [torrust#1264] return wether swarm stats have change or not after upserting a peer (Jose Celano)

Pull request description:

  ### Problem

  **The number of downloads for a torrent is not persisted correctly**.

  Even when persistence is enabled, it does not work. There are two cases when the counter is reset:

  - When you restart the tracker.
  - When a torrent entry is removed from the torrents repository because the last peer was removed (peerless torrents). This only happens if "removing peerless torrents" is enabled in the configuration.

  Check the [issue](torrust#1264) to know how to reproduce the bug.

  ### Solution

  Every time a new torrent entry is added in the torrent repository we need to load stats (also called persistent torrent, which only includes number of downloads currently) from the database. If persistence for stats is enabled.

  ### Other considerations

  **1. Unused code**

  There is a method to load the counters for all torrents but it's not used in production code.

  ```rust
  fn import_persistent(&self, persistent_torrents: &PersistentTorrents) {
      for (info_hash, completed) in persistent_torrents {
          if self.torrents.contains_key(info_hash) {
              continue;
          }

          let entry = EntryMutexStd::new(
              EntrySingle {
                  swarm: PeerList::default(),
                  downloaded: *completed,
              }
              .into(),
          );

          self.torrents.insert(*info_hash, entry);
      }
  }
  ```

  I guess it was added to load the information from the database when the tracker is restarted. However that's not enough because torrents are removed from the repository when they do not have peers (peerless torrents). That would work only if removing peerless torrents is disabled.

  **2. Better solution**

  I think there is a better solution than the one this PR applies. This PR is a temporary patch to fix the bug asap.

  I would differentiate to different counters:

  1. number of downloads per torrent while it was active in the swarm (since the tracker was restarted)
  2. historical number of downloads per torrent per web service (since the service was put into production)

  The counter 1 is what we have before merging this PR. The counter is reset every time the torrent is removed from the swarm. We can keep like it's now before merging this PR. This counter is handled at the in-memory torrent repository.

  The new counter (counter 2) would be handled by the announce handler. After processing the announce request if the number of downloads has increased the handler also increases the historical counter. This new value should be the one exposed in the API.

  ```json
  {
    "torrents": 379508,
    "seeders": 150440,
    "completed": 58992,  <- this should be the historical
    "leechers": 264865,
    "tcp4_connections_handled": 15,
    "tcp4_announces_handled": 15,
    "tcp4_scrapes_handled": 0,
    "tcp6_connections_handled": 0,
    "tcp6_announces_handled": 0,
    "tcp6_scrapes_handled": 0,
    "udp_requests_aborted": 556458,
    "udp_requests_banned": 24773536,
    "udp_banned_ips_total": 29868,
    "udp_avg_connect_processing_time_ns": 478212,
    "udp_avg_announce_processing_time_ns": 8724637,
    "udp_avg_scrape_processing_time_ns": 296950,
    "udp4_requests": 298728340,
    "udp4_connections_handled": 104369821,
    "udp4_announces_handled": 158354656,
    "udp4_scrapes_handled": 3664492,
    "udp4_responses": 273954123,
    "udp4_errors_handled": 7540716,
    "udp6_requests": 0,
    "udp6_connections_handled": 0,
    "udp6_announces_handled": 0,
    "udp6_scrapes_handled": 0,
    "udp6_responses": 0,
    "udp6_errors_handled": 0
  }
  ```

  This patches solves the problem but pre-loading the historical data every time the torrent is added to the repository.

ACKs for top commit:
  josecelano:
    ACK 06ef1d5

Tree-SHA512: 2d1ad14d26efa981d24b1264640b5e2deaca621092ec61638e85e21dc83f92ad53a4041cb97e4e798611f3bfed610c7794b68027e43a847751c8ec022abc6b19

Author: josecelano

packages/axum-http-tracker-server/src/environment.rs

@@ -22,11 +22,11 @@ pub struct Environment<S> {
 impl<S> Environment<S> {
     /// Add a torrent to the tracker
     pub fn add_torrent_peer(&self, info_hash: &InfoHash, peer: &peer::Peer) {
-        let () = self
+        let _number_of_downloads_increased = self
             .container
             .tracker_core_container
             .in_memory_torrent_repository
-            .upsert_peer(info_hash, peer);
+            .upsert_peer(info_hash, peer, None);
     }
 }
 

packages/axum-rest-tracker-api-server/src/environment.rs

@@ -33,11 +33,11 @@ where
 {
     /// Add a torrent to the tracker
     pub fn add_torrent_peer(&self, info_hash: &InfoHash, peer: &peer::Peer) {
-        let () = self
+        let _number_of_downloads_increased = self
             .container
             .tracker_core_container
             .in_memory_torrent_repository
-            .upsert_peer(info_hash, peer);
+            .upsert_peer(info_hash, peer, None);
     }
 }
 

packages/torrent-repository/benches/helpers/asyn.rs

@@ -18,7 +18,7 @@ where
 
         let info_hash = InfoHash::default();
 
-        torrent_repository.upsert_peer(&info_hash, &DEFAULT_PEER).await;
+        torrent_repository.upsert_peer(&info_hash, &DEFAULT_PEER, None).await;
 
         torrent_repository.get_swarm_metadata(&info_hash).await;
     }
@@ -37,7 +37,7 @@ where
     let handles = FuturesUnordered::new();
 
     // Add the torrent/peer to the torrent repository
-    torrent_repository.upsert_peer(&info_hash, &DEFAULT_PEER).await;
+    torrent_repository.upsert_peer(&info_hash, &DEFAULT_PEER, None).await;
 
     torrent_repository.get_swarm_metadata(&info_hash).await;
 
@@ -47,7 +47,7 @@ where
         let torrent_repository_clone = torrent_repository.clone();
 
         let handle = runtime.spawn(async move {
-            torrent_repository_clone.upsert_peer(&info_hash, &DEFAULT_PEER).await;
+            torrent_repository_clone.upsert_peer(&info_hash, &DEFAULT_PEER, None).await;
 
             torrent_repository_clone.get_swarm_metadata(&info_hash).await;
 
@@ -87,7 +87,7 @@ where
         let torrent_repository_clone = torrent_repository.clone();
 
         let handle = runtime.spawn(async move {
-            torrent_repository_clone.upsert_peer(&info_hash, &DEFAULT_PEER).await;
+            torrent_repository_clone.upsert_peer(&info_hash, &DEFAULT_PEER, None).await;
 
             torrent_repository_clone.get_swarm_metadata(&info_hash).await;
 
@@ -123,7 +123,7 @@ where
 
     // Add the torrents/peers to the torrent repository
     for info_hash in &info_hashes {
-        torrent_repository.upsert_peer(info_hash, &DEFAULT_PEER).await;
+        torrent_repository.upsert_peer(info_hash, &DEFAULT_PEER, None).await;
         torrent_repository.get_swarm_metadata(info_hash).await;
     }
 
@@ -133,7 +133,7 @@ where
         let torrent_repository_clone = torrent_repository.clone();
 
         let handle = runtime.spawn(async move {
-            torrent_repository_clone.upsert_peer(&info_hash, &DEFAULT_PEER).await;
+            torrent_repository_clone.upsert_peer(&info_hash, &DEFAULT_PEER, None).await;
             torrent_repository_clone.get_swarm_metadata(&info_hash).await;
 
             if let Some(sleep_time) = sleep {

packages/torrent-repository/benches/helpers/sync.rs

@@ -20,7 +20,7 @@ where
 
         let info_hash = InfoHash::default();
 
-        torrent_repository.upsert_peer(&info_hash, &DEFAULT_PEER);
+        torrent_repository.upsert_peer(&info_hash, &DEFAULT_PEER, None);
 
         torrent_repository.get_swarm_metadata(&info_hash);
     }
@@ -39,7 +39,7 @@ where
     let handles = FuturesUnordered::new();
 
     // Add the torrent/peer to the torrent repository
-    torrent_repository.upsert_peer(&info_hash, &DEFAULT_PEER);
+    torrent_repository.upsert_peer(&info_hash, &DEFAULT_PEER, None);
 
     torrent_repository.get_swarm_metadata(&info_hash);
 
@@ -49,7 +49,7 @@ where
         let torrent_repository_clone = torrent_repository.clone();
 
         let handle = runtime.spawn(async move {
-            torrent_repository_clone.upsert_peer(&info_hash, &DEFAULT_PEER);
+            torrent_repository_clone.upsert_peer(&info_hash, &DEFAULT_PEER, None);
 
             torrent_repository_clone.get_swarm_metadata(&info_hash);
 
@@ -89,7 +89,7 @@ where
         let torrent_repository_clone = torrent_repository.clone();
 
         let handle = runtime.spawn(async move {
-            torrent_repository_clone.upsert_peer(&info_hash, &DEFAULT_PEER);
+            torrent_repository_clone.upsert_peer(&info_hash, &DEFAULT_PEER, None);
 
             torrent_repository_clone.get_swarm_metadata(&info_hash);
 
@@ -125,7 +125,7 @@ where
 
     // Add the torrents/peers to the torrent repository
     for info_hash in &info_hashes {
-        torrent_repository.upsert_peer(info_hash, &DEFAULT_PEER);
+        torrent_repository.upsert_peer(info_hash, &DEFAULT_PEER, None);
         torrent_repository.get_swarm_metadata(info_hash);
     }
 
@@ -135,7 +135,7 @@ where
         let torrent_repository_clone = torrent_repository.clone();
 
         let handle = runtime.spawn(async move {
-            torrent_repository_clone.upsert_peer(&info_hash, &DEFAULT_PEER);
+            torrent_repository_clone.upsert_peer(&info_hash, &DEFAULT_PEER, None);
             torrent_repository_clone.get_swarm_metadata(&info_hash);
 
             if let Some(sleep_time) = sleep {

packages/torrent-repository/src/entry/single.rs

@@ -51,7 +51,7 @@ impl Entry for EntrySingle {
     }
 
     fn upsert_peer(&mut self, peer: &peer::Peer) -> bool {
-        let mut downloaded_stats_updated: bool = false;
+        let mut number_of_downloads_increased: bool = false;
 
         match peer::ReadInfo::get_event(peer) {
             AnnounceEvent::Stopped => {
@@ -62,15 +62,17 @@ impl Entry for EntrySingle {
                 // Don't count if peer was not previously known and not already completed.
                 if previous.is_some_and(|p| p.event != AnnounceEvent::Completed) {
                     self.downloaded += 1;
-                    downloaded_stats_updated = true;
+                    number_of_downloads_increased = true;
                 }
             }
             _ => {
+                // `Started` event (first announced event) or
+                // `None` event (announcements done at regular intervals).
                 drop(self.swarm.upsert(Arc::new(*peer)));
             }
         }
 
-        downloaded_stats_updated
+        number_of_downloads_increased
     }
 
     fn remove_inactive_peers(&mut self, current_cutoff: DurationSinceUnixEpoch) {

packages/torrent-repository/src/repository/dash_map_mutex_std.rs

@@ -6,7 +6,7 @@ use torrust_tracker_configuration::TrackerPolicy;
 use torrust_tracker_primitives::pagination::Pagination;
 use torrust_tracker_primitives::swarm_metadata::SwarmMetadata;
 use torrust_tracker_primitives::torrent_metrics::TorrentsMetrics;
-use torrust_tracker_primitives::{peer, DurationSinceUnixEpoch, PersistentTorrents};
+use torrust_tracker_primitives::{peer, DurationSinceUnixEpoch, PersistentTorrent, PersistentTorrents};
 
 use super::Repository;
 use crate::entry::peer_list::PeerList;
@@ -23,13 +23,17 @@ where
     EntryMutexStd: EntrySync,
     EntrySingle: Entry,
 {
-    fn upsert_peer(&self, info_hash: &InfoHash, peer: &peer::Peer) {
+    fn upsert_peer(&self, info_hash: &InfoHash, peer: &peer::Peer, _opt_persistent_torrent: Option<PersistentTorrent>) -> bool {
+        // todo: load persistent torrent data if provided
+
         if let Some(entry) = self.torrents.get(info_hash) {
-            entry.upsert_peer(peer);
+            entry.upsert_peer(peer)
         } else {
             let _unused = self.torrents.insert(*info_hash, Arc::default());
             if let Some(entry) = self.torrents.get(info_hash) {
-                entry.upsert_peer(peer);
+                entry.upsert_peer(peer)
+            } else {
+                false
             }
         }
     }

packages/torrent-repository/src/repository/mod.rs

@@ -3,7 +3,7 @@ use torrust_tracker_configuration::TrackerPolicy;
 use torrust_tracker_primitives::pagination::Pagination;
 use torrust_tracker_primitives::swarm_metadata::SwarmMetadata;
 use torrust_tracker_primitives::torrent_metrics::TorrentsMetrics;
-use torrust_tracker_primitives::{peer, DurationSinceUnixEpoch, PersistentTorrents};
+use torrust_tracker_primitives::{peer, DurationSinceUnixEpoch, PersistentTorrent, PersistentTorrents};
 
 pub mod dash_map_mutex_std;
 pub mod rw_lock_std;
@@ -24,7 +24,7 @@ pub trait Repository<T>: Debug + Default + Sized + 'static {
     fn remove(&self, key: &InfoHash) -> Option<T>;
     fn remove_inactive_peers(&self, current_cutoff: DurationSinceUnixEpoch);
     fn remove_peerless_torrents(&self, policy: &TrackerPolicy);
-    fn upsert_peer(&self, info_hash: &InfoHash, peer: &peer::Peer);
+    fn upsert_peer(&self, info_hash: &InfoHash, peer: &peer::Peer, opt_persistent_torrent: Option<PersistentTorrent>) -> bool;
     fn get_swarm_metadata(&self, info_hash: &InfoHash) -> Option<SwarmMetadata>;
 }
 
@@ -37,6 +37,11 @@ pub trait RepositoryAsync<T>: Debug + Default + Sized + 'static {
     fn remove(&self, key: &InfoHash) -> impl std::future::Future<Output = Option<T>> + Send;
     fn remove_inactive_peers(&self, current_cutoff: DurationSinceUnixEpoch) -> impl std::future::Future<Output = ()> + Send;
     fn remove_peerless_torrents(&self, policy: &TrackerPolicy) -> impl std::future::Future<Output = ()> + Send;
-    fn upsert_peer(&self, info_hash: &InfoHash, peer: &peer::Peer) -> impl std::future::Future<Output = ()> + Send;
+    fn upsert_peer(
+        &self,
+        info_hash: &InfoHash,
+        peer: &peer::Peer,
+        opt_persistent_torrent: Option<PersistentTorrent>,
+    ) -> impl std::future::Future<Output = bool> + Send;
     fn get_swarm_metadata(&self, info_hash: &InfoHash) -> impl std::future::Future<Output = Option<SwarmMetadata>> + Send;
 }

packages/torrent-repository/src/repository/rw_lock_std.rs

@@ -3,7 +3,7 @@ use torrust_tracker_configuration::TrackerPolicy;
 use torrust_tracker_primitives::pagination::Pagination;
 use torrust_tracker_primitives::swarm_metadata::SwarmMetadata;
 use torrust_tracker_primitives::torrent_metrics::TorrentsMetrics;
-use torrust_tracker_primitives::{peer, DurationSinceUnixEpoch, PersistentTorrents};
+use torrust_tracker_primitives::{peer, DurationSinceUnixEpoch, PersistentTorrent, PersistentTorrents};
 
 use super::Repository;
 use crate::entry::peer_list::PeerList;
@@ -46,12 +46,14 @@ impl Repository<EntrySingle> for TorrentsRwLockStd
 where
     EntrySingle: Entry,
 {
-    fn upsert_peer(&self, info_hash: &InfoHash, peer: &peer::Peer) {
+    fn upsert_peer(&self, info_hash: &InfoHash, peer: &peer::Peer, _opt_persistent_torrent: Option<PersistentTorrent>) -> bool {
+        // todo: load persistent torrent data if provided
+
         let mut db = self.get_torrents_mut();
 
         let entry = db.entry(*info_hash).or_insert(EntrySingle::default());
 
-        entry.upsert_peer(peer);
+        entry.upsert_peer(peer)
     }
 
     fn get_swarm_metadata(&self, info_hash: &InfoHash) -> Option<SwarmMetadata> {

packages/torrent-repository/src/repository/rw_lock_std_mutex_std.rs

@@ -5,7 +5,7 @@ use torrust_tracker_configuration::TrackerPolicy;
 use torrust_tracker_primitives::pagination::Pagination;
 use torrust_tracker_primitives::swarm_metadata::SwarmMetadata;
 use torrust_tracker_primitives::torrent_metrics::TorrentsMetrics;
-use torrust_tracker_primitives::{peer, DurationSinceUnixEpoch, PersistentTorrents};
+use torrust_tracker_primitives::{peer, DurationSinceUnixEpoch, PersistentTorrent, PersistentTorrents};
 
 use super::Repository;
 use crate::entry::peer_list::PeerList;
@@ -33,7 +33,9 @@ where
     EntryMutexStd: EntrySync,
     EntrySingle: Entry,
 {
-    fn upsert_peer(&self, info_hash: &InfoHash, peer: &peer::Peer) {
+    fn upsert_peer(&self, info_hash: &InfoHash, peer: &peer::Peer, _opt_persistent_torrent: Option<PersistentTorrent>) -> bool {
+        // todo: load persistent torrent data if provided
+
         let maybe_entry = self.get_torrents().get(info_hash).cloned();
 
         let entry = if let Some(entry) = maybe_entry {
@@ -44,7 +46,7 @@ where
             entry.clone()
         };
 
-        entry.upsert_peer(peer);
+        entry.upsert_peer(peer)
     }
 
     fn get_swarm_metadata(&self, info_hash: &InfoHash) -> Option<SwarmMetadata> {

packages/torrent-repository/src/repository/rw_lock_std_mutex_tokio.rs

@@ -9,7 +9,7 @@ use torrust_tracker_configuration::TrackerPolicy;
 use torrust_tracker_primitives::pagination::Pagination;
 use torrust_tracker_primitives::swarm_metadata::SwarmMetadata;
 use torrust_tracker_primitives::torrent_metrics::TorrentsMetrics;
-use torrust_tracker_primitives::{peer, DurationSinceUnixEpoch, PersistentTorrents};
+use torrust_tracker_primitives::{peer, DurationSinceUnixEpoch, PersistentTorrent, PersistentTorrents};
 
 use super::RepositoryAsync;
 use crate::entry::peer_list::PeerList;
@@ -37,7 +37,14 @@ where
     EntryMutexTokio: EntryAsync,
     EntrySingle: Entry,
 {
-    async fn upsert_peer(&self, info_hash: &InfoHash, peer: &peer::Peer) {
+    async fn upsert_peer(
+        &self,
+        info_hash: &InfoHash,
+        peer: &peer::Peer,
+        _opt_persistent_torrent: Option<PersistentTorrent>,
+    ) -> bool {
+        // todo: load persistent torrent data if provided
+
         let maybe_entry = self.get_torrents().get(info_hash).cloned();
 
         let entry = if let Some(entry) = maybe_entry {
@@ -48,7 +55,7 @@ where
             entry.clone()
         };
 
-        entry.upsert_peer(peer).await;
+        entry.upsert_peer(peer).await
     }
 
     async fn get_swarm_metadata(&self, info_hash: &InfoHash) -> Option<SwarmMetadata> {

packages/torrent-repository/src/repository/rw_lock_tokio.rs

@@ -3,7 +3,7 @@ use torrust_tracker_configuration::TrackerPolicy;
 use torrust_tracker_primitives::pagination::Pagination;
 use torrust_tracker_primitives::swarm_metadata::SwarmMetadata;
 use torrust_tracker_primitives::torrent_metrics::TorrentsMetrics;
-use torrust_tracker_primitives::{peer, DurationSinceUnixEpoch, PersistentTorrents};
+use torrust_tracker_primitives::{peer, DurationSinceUnixEpoch, PersistentTorrent, PersistentTorrents};
 
 use super::RepositoryAsync;
 use crate::entry::peer_list::PeerList;
@@ -47,12 +47,19 @@ impl RepositoryAsync<EntrySingle> for TorrentsRwLockTokio
 where
     EntrySingle: Entry,
 {
-    async fn upsert_peer(&self, info_hash: &InfoHash, peer: &peer::Peer) {
+    async fn upsert_peer(
+        &self,
+        info_hash: &InfoHash,
+        peer: &peer::Peer,
+        _opt_persistent_torrent: Option<PersistentTorrent>,
+    ) -> bool {
+        // todo: load persistent torrent data if provided
+
         let mut db = self.get_torrents_mut().await;
 
         let entry = db.entry(*info_hash).or_insert(EntrySingle::default());
 
-        entry.upsert_peer(peer);
+        entry.upsert_peer(peer)
     }
 
     async fn get_swarm_metadata(&self, info_hash: &InfoHash) -> Option<SwarmMetadata> {

packages/torrent-repository/src/repository/rw_lock_tokio_mutex_std.rs

@@ -5,7 +5,7 @@ use torrust_tracker_configuration::TrackerPolicy;
 use torrust_tracker_primitives::pagination::Pagination;
 use torrust_tracker_primitives::swarm_metadata::SwarmMetadata;
 use torrust_tracker_primitives::torrent_metrics::TorrentsMetrics;
-use torrust_tracker_primitives::{peer, DurationSinceUnixEpoch, PersistentTorrents};
+use torrust_tracker_primitives::{peer, DurationSinceUnixEpoch, PersistentTorrent, PersistentTorrents};
 
 use super::RepositoryAsync;
 use crate::entry::peer_list::PeerList;
@@ -35,7 +35,14 @@ where
     EntryMutexStd: EntrySync,
     EntrySingle: Entry,
 {
-    async fn upsert_peer(&self, info_hash: &InfoHash, peer: &peer::Peer) {
+    async fn upsert_peer(
+        &self,
+        info_hash: &InfoHash,
+        peer: &peer::Peer,
+        _opt_persistent_torrent: Option<PersistentTorrent>,
+    ) -> bool {
+        // todo: load persistent torrent data if provided
+
         let maybe_entry = self.get_torrents().await.get(info_hash).cloned();
 
         let entry = if let Some(entry) = maybe_entry {
@@ -46,7 +53,7 @@ where
             entry.clone()
         };
 
-        entry.upsert_peer(peer);
+        entry.upsert_peer(peer)
     }
 
     async fn get_swarm_metadata(&self, info_hash: &InfoHash) -> Option<SwarmMetadata> {