feat: [#1539] load global downloads counter from DB

When the tracker starts.

Author: josecelano

packages/metrics/src/counter.rs

@@ -20,6 +20,10 @@ impl Counter {
     pub fn increment(&mut self, value: u64) {
         self.0 += value;
     }
+
+    pub fn absolute(&mut self, value: u64) {
+        self.0 = value;
+    }
 }
 
 impl From<u64> for Counter {
@@ -73,6 +77,13 @@ mod tests {
         assert_eq!(counter.value(), 3);
     }
 
+    #[test]
+    fn it_could_set_to_an_absolute_value() {
+        let mut counter = Counter::new(0);
+        counter.absolute(1);
+        assert_eq!(counter.value(), 1);
+    }
+
     #[test]
     fn it_serializes_to_prometheus() {
         let counter = Counter::new(42);

packages/metrics/src/metric/mod.rs

@@ -55,6 +55,10 @@ impl Metric<Counter> {
     pub fn increment(&mut self, label_set: &LabelSet, time: DurationSinceUnixEpoch) {
         self.sample_collection.increment(label_set, time);
     }
+
+    pub fn absolute(&mut self, label_set: &LabelSet, value: u64, time: DurationSinceUnixEpoch) {
+        self.sample_collection.absolute(label_set, value, time);
+    }
 }
 
 impl Metric<Gauge> {

packages/metrics/src/metric_collection.rs

@@ -72,6 +72,8 @@ impl MetricCollection {
         self.counters.get_value(name, label_set)
     }
 
+    /// Increases the counter for the given metric name and labels.
+    ///
     /// # Errors
     ///
     /// Return an error if a metrics of a different type with the same name
@@ -93,6 +95,30 @@ impl MetricCollection {
         Ok(())
     }
 
+    /// Sets the counter for the given metric name and labels.
+    ///
+    /// # Errors
+    ///
+    /// Return an error if a metrics of a different type with the same name
+    /// already exists.
+    pub fn set_counter(
+        &mut self,
+        name: &MetricName,
+        label_set: &LabelSet,
+        value: u64,
+        time: DurationSinceUnixEpoch,
+    ) -> Result<(), Error> {
+        if self.gauges.metrics.contains_key(name) {
+            return Err(Error::MetricNameCollisionAdding {
+                metric_name: name.clone(),
+            });
+        }
+
+        self.counters.absolute(name, label_set, value, time);
+
+        Ok(())
+    }
+
     pub fn ensure_counter_exists(&mut self, name: &MetricName) {
         self.counters.ensure_metric_exists(name);
     }
@@ -361,7 +387,7 @@ impl MetricKindCollection<Counter> {
     ///
     /// # Panics
     ///
-    /// Panics if the metric does not exist and it could not be created.
+    /// Panics if the metric does not exist.
     pub fn increment(&mut self, name: &MetricName, label_set: &LabelSet, time: DurationSinceUnixEpoch) {
         self.ensure_metric_exists(name);
 
@@ -370,6 +396,21 @@ impl MetricKindCollection<Counter> {
         metric.increment(label_set, time);
     }
 
+    /// Sets the counter to an absolute value for the given metric name and labels.
+    ///
+    /// If the metric name does not exist, it will be created.
+    ///
+    /// # Panics
+    ///
+    /// Panics if the metric does not exist.
+    pub fn absolute(&mut self, name: &MetricName, label_set: &LabelSet, value: u64, time: DurationSinceUnixEpoch) {
+        self.ensure_metric_exists(name);
+
+        let metric = self.metrics.get_mut(name).expect("Counter metric should exist");
+
+        metric.absolute(label_set, value, time);
+    }
+
     #[must_use]
     pub fn get_value(&self, name: &MetricName, label_set: &LabelSet) -> Option<Counter> {
         self.metrics

packages/metrics/src/sample.rs

@@ -122,6 +122,11 @@ impl Measurement<Counter> {
         self.value.increment(1);
         self.set_recorded_at(time);
     }
+
+    pub fn absolute(&mut self, value: u64, time: DurationSinceUnixEpoch) {
+        self.value.absolute(value);
+        self.set_recorded_at(time);
+    }
 }
 
 impl Measurement<Gauge> {

packages/metrics/src/sample_collection.rs

@@ -79,6 +79,15 @@ impl SampleCollection<Counter> {
 
         sample.increment(time);
     }
+
+    pub fn absolute(&mut self, label_set: &LabelSet, value: u64, time: DurationSinceUnixEpoch) {
+        let sample = self
+            .samples
+            .entry(label_set.clone())
+            .or_insert_with(|| Measurement::new(Counter::default(), time));
+
+        sample.absolute(value, time);
+    }
 }
 
 impl SampleCollection<Gauge> {

packages/tracker-core/src/statistics/metrics.rs

@@ -24,6 +24,19 @@ impl Metrics {
         self.metric_collection.increase_counter(metric_name, labels, now)
     }
 
+    /// # Errors
+    ///
+    /// Returns an error if the metric does not exist and it cannot be created.
+    pub fn set_counter(
+        &mut self,
+        metric_name: &MetricName,
+        labels: &LabelSet,
+        value: u64,
+        now: DurationSinceUnixEpoch,
+    ) -> Result<(), Error> {
+        self.metric_collection.set_counter(metric_name, labels, value, now)
+    }
+
     /// # Errors
     ///
     /// Returns an error if the metric does not exist and it cannot be created.

packages/tracker-core/src/statistics/mod.rs

@@ -1,5 +1,6 @@
 pub mod event;
 pub mod metrics;
+pub mod persisted_metrics;
 pub mod repository;
 
 use metrics::Metrics;

packages/tracker-core/src/statistics/persisted_metrics.rs

@@ -0,0 +1,57 @@
+use std::sync::Arc;
+
+use thiserror::Error;
+use torrust_tracker_metrics::label::LabelSet;
+use torrust_tracker_metrics::{metric_collection, metric_name};
+use torrust_tracker_primitives::DurationSinceUnixEpoch;
+
+use super::repository::Repository;
+use super::TRACKER_CORE_PERSISTENT_TORRENTS_DOWNLOADS_TOTAL;
+use crate::databases;
+use crate::torrent::repository::persisted::DatabasePersistentTorrentRepository;
+
+/// Loads persisted metrics from the database and sets them in the stats repository.
+///
+/// # Errors
+///
+/// This function will return an error if the database query fails or if the
+/// metric collection fails to set the initial metric values.
+pub async fn load_persisted_metrics(
+    stats_repository: &Arc<Repository>,
+    db_torrent_repository: &Arc<DatabasePersistentTorrentRepository>,
+    now: DurationSinceUnixEpoch,
+) -> Result<(), Error> {
+    if let Some(downloads) = db_torrent_repository.load_global_number_of_downloads()? {
+        stats_repository
+            .set_counter(
+                &metric_name!(TRACKER_CORE_PERSISTENT_TORRENTS_DOWNLOADS_TOTAL),
+                &LabelSet::default(),
+                u64::from(downloads),
+                now,
+            )
+            .await?;
+    }
+
+    Ok(())
+}
+
+#[derive(Error, Debug, Clone)]
+pub enum Error {
+    #[error("Database error: {err}")]
+    DatabaseError { err: databases::error::Error },
+
+    #[error("Metrics error: {err}")]
+    MetricsError { err: metric_collection::Error },
+}
+
+impl From<databases::error::Error> for Error {
+    fn from(err: databases::error::Error) -> Self {
+        Self::DatabaseError { err }
+    }
+}
+
+impl From<metric_collection::Error> for Error {
+    fn from(err: metric_collection::Error) -> Self {
+        Self::MetricsError { err }
+    }
+}

packages/tracker-core/src/statistics/repository.rs

@@ -57,6 +57,31 @@ impl Repository {
         result
     }
 
+    /// # Errors
+    ///
+    /// This function will return an error if the metric collection fails to
+    /// increment the counter.
+    pub async fn set_counter(
+        &self,
+        metric_name: &MetricName,
+        labels: &LabelSet,
+        value: u64,
+        now: DurationSinceUnixEpoch,
+    ) -> Result<(), Error> {
+        let mut stats_lock = self.stats.write().await;
+
+        let result = stats_lock.set_counter(metric_name, labels, value, now);
+
+        drop(stats_lock);
+
+        match result {
+            Ok(()) => {}
+            Err(ref err) => tracing::error!("Failed to set the counter: {}", err),
+        }
+
+        result
+    }
+
     /// # Errors
     ///
     /// This function will return an error if the metric collection fails to

packages/tracker-core/src/torrent/repository/persisted.rs

@@ -47,6 +47,8 @@ impl DatabasePersistentTorrentRepository {
         }
     }
 
+    // Single Torrent Metrics
+
     /// Increases the number of downloads for a given torrent.
     ///
     /// If the torrent is not found, it creates a new entry.
@@ -67,22 +69,6 @@ impl DatabasePersistentTorrentRepository {
         }
     }
 
-    /// Increases the global number of downloads for all torrent.
-    ///
-    /// If the metric is not found, it creates it.
-    ///
-    /// # Errors
-    ///
-    /// Returns an [`Error`] if the database operation fails.
-    pub(crate) fn increase_global_number_of_downloads(&self) -> Result<(), Error> {
-        let torrent = self.database.load_global_number_of_downloads()?;
-
-        match torrent {
-            Some(_number_of_downloads) => self.database.increase_global_number_of_downloads(),
-            None => self.database.save_global_number_of_downloads(1),
-        }
-    }
-
     /// Loads all persistent torrent metrics from the database.
     ///
     /// This function retrieves the torrent metrics (e.g., download counts) from the persistent store
@@ -123,6 +109,33 @@ impl DatabasePersistentTorrentRepository {
     pub(crate) fn save(&self, info_hash: &InfoHash, downloaded: u32) -> Result<(), Error> {
         self.database.save_persistent_torrent(info_hash, downloaded)
     }
+
+    // Aggregate Metrics
+
+    /// Increases the global number of downloads for all torrent.
+    ///
+    /// If the metric is not found, it creates it.
+    ///
+    /// # Errors
+    ///
+    /// Returns an [`Error`] if the database operation fails.
+    pub(crate) fn increase_global_number_of_downloads(&self) -> Result<(), Error> {
+        let torrent = self.database.load_global_number_of_downloads()?;
+
+        match torrent {
+            Some(_number_of_downloads) => self.database.increase_global_number_of_downloads(),
+            None => self.database.save_global_number_of_downloads(1),
+        }
+    }
+
+    /// Loads the global number of downloads for all torrents from the database.
+    ///
+    /// # Errors
+    ///
+    /// Returns an [`Error`] if the underlying database query fails.
+    pub(crate) fn load_global_number_of_downloads(&self) -> Result<Option<PersistentTorrent>, Error> {
+        self.database.load_global_number_of_downloads()
+    }
 }
 
 #[cfg(test)]