fix: [#1589] add dedicated metric for UDP request processing in moving average calculation

Add a new metric `UDP_TRACKER_SERVER_PERFORMANCE_PROCESSED_REQUESTS_TOTAL` to track
requests processed specifically for performance metrics, eliminating race conditions
in the moving average calculation.

**Changes:**
- Add new metric constant `UDP_TRACKER_SERVER_PERFORMANCE_PROCESSED_REQUESTS_TOTAL`
- Update `recalculate_udp_avg_processing_time_ns()` to use dedicated counter instead of accepted requests total
- Add `udp_processed_requests_total()` method to retrieve the new metric value
- Add `increment_udp_processed_requests_total()` helper method
- Update metric descriptions to include the new counter

**Problem Fixed:**
Previously, the moving average calculation used the accepted requests counter that could be
updated independently, causing race conditions where the same request count was used for
multiple calculations. The new implementation increments its own dedicated counter atomically
during the calculation, ensuring consistency.

**Behavior Change:**
The counter now starts at 0 and gets incremented to 1 on the first calculation call,
then uses proper moving average formula for subsequent calls. This eliminates division
by zero issues and provides more accurate moving averages.

**Tests Updated:**
Updated repository tests to reflect the new atomic behavior where the processed requests
counter is managed specifically for moving average calculations.

Fixes race conditions in UDP request processing time metrics while maintaining
backward compatibility of all public APIs.

Author: josecelano

packages/udp-tracker-server/src/statistics/metrics.rs

@@ -9,7 +9,8 @@ use torrust_tracker_metrics::metric_name;
 use torrust_tracker_primitives::DurationSinceUnixEpoch;
 
 use crate::statistics::{
-    UDP_TRACKER_SERVER_ERRORS_TOTAL, UDP_TRACKER_SERVER_IPS_BANNED_TOTAL, UDP_TRACKER_SERVER_PERFORMANCE_AVG_PROCESSING_TIME_NS,
+    UDP_TRACKER_SERVER_ERRORS_TOTAL, UDP_TRACKER_SERVER_IPS_BANNED_TOTAL,
+    UDP_TRACKER_SERVER_PERFORMANCE_AVG_PROCESSED_REQUESTS_TOTAL, UDP_TRACKER_SERVER_PERFORMANCE_AVG_PROCESSING_TIME_NS,
     UDP_TRACKER_SERVER_REQUESTS_ABORTED_TOTAL, UDP_TRACKER_SERVER_REQUESTS_ACCEPTED_TOTAL,
     UDP_TRACKER_SERVER_REQUESTS_BANNED_TOTAL, UDP_TRACKER_SERVER_REQUESTS_RECEIVED_TOTAL,
     UDP_TRACKER_SERVER_RESPONSES_SENT_TOTAL,
@@ -57,26 +58,22 @@ impl Metrics {
         label_set: &LabelSet,
         now: DurationSinceUnixEpoch,
     ) -> f64 {
-        let req_processing_time = req_processing_time.as_nanos() as f64;
-
-        let request_accepted_total = self.udp_request_accepted(label_set) as f64;
+        self.increment_udp_processed_requests_total(label_set, now);
 
+        let processed_requests_total = self.udp_processed_requests_total(label_set) as f64;
         let previous_avg = self.udp_avg_processing_time_ns(label_set);
+        let req_processing_time = req_processing_time.as_nanos() as f64;
 
-        let new_avg = if request_accepted_total == 0.0 {
-            req_processing_time
-        } else {
-            // Moving average: https://en.wikipedia.org/wiki/Moving_average
-            previous_avg as f64 + (req_processing_time - previous_avg as f64) / request_accepted_total
-        };
+        // Moving average: https://en.wikipedia.org/wiki/Moving_average
+        let new_avg = previous_avg as f64 + (req_processing_time - previous_avg as f64) / processed_requests_total;
 
         tracing::debug!(
-            "Recalculated UDP average processing time for labels {}: {} ns (previous: {} ns, req_processing_time: {} ns, request_accepted_total: {})",
+            "Recalculated UDP average processing time for labels {}: {} ns (previous: {} ns, req_processing_time: {} ns, request_processed_total: {})",
             label_set,
             new_avg,
             previous_avg,
             req_processing_time,
-            request_accepted_total
+            processed_requests_total
         );
 
         self.update_udp_avg_processing_time_ns(new_avg, label_set, now);
@@ -105,6 +102,18 @@ impl Metrics {
             .unwrap_or_default() as u64
     }
 
+    #[must_use]
+    #[allow(clippy::cast_sign_loss)]
+    #[allow(clippy::cast_possible_truncation)]
+    pub fn udp_processed_requests_total(&self, label_set: &LabelSet) -> u64 {
+        self.metric_collection
+            .sum(
+                &metric_name!(UDP_TRACKER_SERVER_PERFORMANCE_AVG_PROCESSED_REQUESTS_TOTAL),
+                label_set,
+            )
+            .unwrap_or_default() as u64
+    }
+
     fn update_udp_avg_processing_time_ns(&mut self, new_avg: f64, label_set: &LabelSet, now: DurationSinceUnixEpoch) {
         tracing::debug!(
             "Updating average processing time metric to {} ns for label set {}",
@@ -123,6 +132,19 @@ impl Metrics {
         }
     }
 
+    fn increment_udp_processed_requests_total(&mut self, label_set: &LabelSet, now: DurationSinceUnixEpoch) {
+        tracing::debug!("Incrementing processed requests total for label set {}", label_set,);
+
+        match self.increase_counter(
+            &metric_name!(UDP_TRACKER_SERVER_PERFORMANCE_AVG_PROCESSED_REQUESTS_TOTAL),
+            label_set,
+            now,
+        ) {
+            Ok(()) => {}
+            Err(err) => tracing::error!("Failed to increment counter: {}", err),
+        }
+    }
+
     // UDP
     /// Total number of UDP (UDP tracker) requests aborted.
     #[must_use]
@@ -360,9 +382,10 @@ mod tests {
     use super::*;
     use crate::statistics::{
         UDP_TRACKER_SERVER_ERRORS_TOTAL, UDP_TRACKER_SERVER_IPS_BANNED_TOTAL,
-        UDP_TRACKER_SERVER_PERFORMANCE_AVG_PROCESSING_TIME_NS, UDP_TRACKER_SERVER_REQUESTS_ABORTED_TOTAL,
-        UDP_TRACKER_SERVER_REQUESTS_ACCEPTED_TOTAL, UDP_TRACKER_SERVER_REQUESTS_BANNED_TOTAL,
-        UDP_TRACKER_SERVER_REQUESTS_RECEIVED_TOTAL, UDP_TRACKER_SERVER_RESPONSES_SENT_TOTAL,
+        UDP_TRACKER_SERVER_PERFORMANCE_AVG_PROCESSED_REQUESTS_TOTAL, UDP_TRACKER_SERVER_PERFORMANCE_AVG_PROCESSING_TIME_NS,
+        UDP_TRACKER_SERVER_REQUESTS_ABORTED_TOTAL, UDP_TRACKER_SERVER_REQUESTS_ACCEPTED_TOTAL,
+        UDP_TRACKER_SERVER_REQUESTS_BANNED_TOTAL, UDP_TRACKER_SERVER_REQUESTS_RECEIVED_TOTAL,
+        UDP_TRACKER_SERVER_RESPONSES_SENT_TOTAL,
     };
     use crate::CurrentClock;
 
@@ -437,6 +460,31 @@ mod tests {
         assert!(result.is_ok());
     }
 
+    #[test]
+    fn it_should_return_zero_for_udp_processed_requests_total_when_no_data() {
+        let metrics = Metrics::default();
+        let labels = LabelSet::from([("request_kind", "connect")]);
+        assert_eq!(metrics.udp_processed_requests_total(&labels), 0);
+    }
+
+    #[test]
+    fn it_should_increment_processed_requests_total() {
+        let mut metrics = Metrics::default();
+        let now = CurrentClock::now();
+        let labels = LabelSet::from([("request_kind", "connect")]);
+
+        // Directly increment the counter using the public method
+        metrics
+            .increase_counter(
+                &metric_name!(UDP_TRACKER_SERVER_PERFORMANCE_AVG_PROCESSED_REQUESTS_TOTAL),
+                &labels,
+                now,
+            )
+            .unwrap();
+
+        assert_eq!(metrics.udp_processed_requests_total(&labels), 1);
+    }
+
     mod udp_general_metrics {
         use super::*;
 

packages/udp-tracker-server/src/statistics/mod.rs

@@ -17,6 +17,8 @@ pub const UDP_TRACKER_SERVER_REQUESTS_ACCEPTED_TOTAL: &str = "udp_tracker_server
 pub const UDP_TRACKER_SERVER_RESPONSES_SENT_TOTAL: &str = "udp_tracker_server_responses_sent_total";
 pub const UDP_TRACKER_SERVER_ERRORS_TOTAL: &str = "udp_tracker_server_errors_total";
 pub const UDP_TRACKER_SERVER_PERFORMANCE_AVG_PROCESSING_TIME_NS: &str = "udp_tracker_server_performance_avg_processing_time_ns";
+pub const UDP_TRACKER_SERVER_PERFORMANCE_AVG_PROCESSED_REQUESTS_TOTAL: &str =
+    "udp_tracker_server_performance_avg_processed_requests_total";
 
 #[must_use]
 pub fn describe_metrics() -> Metrics {
@@ -76,5 +78,13 @@ pub fn describe_metrics() -> Metrics {
         Some(MetricDescription::new("Average time to process a UDP request in nanoseconds")),
     );
 
+    metrics.metric_collection.describe_counter(
+        &metric_name!(UDP_TRACKER_SERVER_PERFORMANCE_AVG_PROCESSED_REQUESTS_TOTAL),
+        Some(Unit::Count),
+        Some(MetricDescription::new(
+            "Total number of UDP requests processed for the average performance metrics",
+        )),
+    );
+
     metrics
 }

packages/udp-tracker-server/src/statistics/repository.rs

@@ -295,21 +295,6 @@ mod tests {
         let repo = Repository::new();
         let now = CurrentClock::now();
 
-        // Set up initial connections handled
-        let ipv4_labels = LabelSet::from([("server_binding_address_ip_family", "inet"), ("request_kind", "connect")]);
-        let ipv6_labels = LabelSet::from([("server_binding_address_ip_family", "inet6"), ("request_kind", "connect")]);
-
-        // Simulate 2 IPv4 and 1 IPv6 connections
-        repo.increase_counter(&metric_name!(UDP_TRACKER_SERVER_REQUESTS_ACCEPTED_TOTAL), &ipv4_labels, now)
-            .await
-            .unwrap();
-        repo.increase_counter(&metric_name!(UDP_TRACKER_SERVER_REQUESTS_ACCEPTED_TOTAL), &ipv4_labels, now)
-            .await
-            .unwrap();
-        repo.increase_counter(&metric_name!(UDP_TRACKER_SERVER_REQUESTS_ACCEPTED_TOTAL), &ipv6_labels, now)
-            .await
-            .unwrap();
-
         // Set initial average to 1000ns
         let connect_labels = LabelSet::from([("request_kind", "connect")]);
         repo.set_gauge(
@@ -322,14 +307,16 @@ mod tests {
         .unwrap();
 
         // Calculate new average with processing time of 2000ns
+        // This will increment the processed requests counter from 0 to 1
         let processing_time = Duration::from_nanos(2000);
         let new_avg = repo
             .recalculate_udp_avg_processing_time_ns(processing_time, &connect_labels, now)
             .await;
 
-        // Moving average: previous_avg + (new_value - previous_avg) / total_connections
-        // 1000 + (2000 - 1000) / 3 = 1000 + 333.33 = 1333.33
-        let expected_avg = 1000.0 + (2000.0 - 1000.0) / 3.0;
+        // Moving average: previous_avg + (new_value - previous_avg) / processed_requests_total
+        // With processed_requests_total = 1 (incremented during the call):
+        // 1000 + (2000 - 1000) / 1 = 1000 + 1000 = 2000
+        let expected_avg = 1000.0 + (2000.0 - 1000.0) / 1.0;
         assert!(
             (new_avg - expected_avg).abs() < 0.01,
             "Expected {expected_avg}, got {new_avg}"
@@ -341,22 +328,6 @@ mod tests {
         let repo = Repository::new();
         let now = CurrentClock::now();
 
-        // Set up initial announces handled
-        let ipv4_labels = LabelSet::from([("server_binding_address_ip_family", "inet"), ("request_kind", "announce")]);
-        let ipv6_labels = LabelSet::from([("server_binding_address_ip_family", "inet6"), ("request_kind", "announce")]);
-
-        // Simulate 3 IPv4 and 2 IPv6 announces
-        for _ in 0..3 {
-            repo.increase_counter(&metric_name!(UDP_TRACKER_SERVER_REQUESTS_ACCEPTED_TOTAL), &ipv4_labels, now)
-                .await
-                .unwrap();
-        }
-        for _ in 0..2 {
-            repo.increase_counter(&metric_name!(UDP_TRACKER_SERVER_REQUESTS_ACCEPTED_TOTAL), &ipv6_labels, now)
-                .await
-                .unwrap();
-        }
-
         // Set initial average to 500ns
         let announce_labels = LabelSet::from([("request_kind", "announce")]);
         repo.set_gauge(
@@ -369,14 +340,16 @@ mod tests {
         .unwrap();
 
         // Calculate new average with processing time of 1500ns
+        // This will increment the processed requests counter from 0 to 1
         let processing_time = Duration::from_nanos(1500);
         let new_avg = repo
             .recalculate_udp_avg_processing_time_ns(processing_time, &announce_labels, now)
             .await;
 
-        // Moving average: previous_avg + (new_value - previous_avg) / total_announces
-        // 500 + (1500 - 500) / 5 = 500 + 200 = 700
-        let expected_avg = 500.0 + (1500.0 - 500.0) / 5.0;
+        // Moving average: previous_avg + (new_value - previous_avg) / processed_requests_total
+        // With processed_requests_total = 1 (incremented during the call):
+        // 500 + (1500 - 500) / 1 = 500 + 1000 = 1500
+        let expected_avg = 500.0 + (1500.0 - 500.0) / 1.0;
         assert!(
             (new_avg - expected_avg).abs() < 0.01,
             "Expected {expected_avg}, got {new_avg}"
@@ -388,16 +361,6 @@ mod tests {
         let repo = Repository::new();
         let now = CurrentClock::now();
 
-        // Set up initial scrapes handled
-        let ipv4_labels = LabelSet::from([("server_binding_address_ip_family", "inet"), ("request_kind", "scrape")]);
-
-        // Simulate 4 IPv4 scrapes
-        for _ in 0..4 {
-            repo.increase_counter(&metric_name!(UDP_TRACKER_SERVER_REQUESTS_ACCEPTED_TOTAL), &ipv4_labels, now)
-                .await
-                .unwrap();
-        }
-
         // Set initial average to 800ns
         let scrape_labels = LabelSet::from([("request_kind", "scrape")]);
         repo.set_gauge(
@@ -410,14 +373,16 @@ mod tests {
         .unwrap();
 
         // Calculate new average with processing time of 1200ns
+        // This will increment the processed requests counter from 0 to 1
         let processing_time = Duration::from_nanos(1200);
         let new_avg = repo
             .recalculate_udp_avg_processing_time_ns(processing_time, &scrape_labels, now)
             .await;
 
-        // Moving average: previous_avg + (new_value - previous_avg) / total_scrapes
-        // 800 + (1200 - 800) / 4 = 800 + 100 = 900
-        let expected_avg = 800.0 + (1200.0 - 800.0) / 4.0;
+        // Moving average: previous_avg + (new_value - previous_avg) / processed_requests_total
+        // With processed_requests_total = 1 (incremented during the call):
+        // 800 + (1200 - 800) / 1 = 800 + 400 = 1200
+        let expected_avg = 800.0 + (1200.0 - 800.0) / 1.0;
         assert!(
             (new_avg - expected_avg).abs() < 0.01,
             "Expected {expected_avg}, got {new_avg}"
@@ -584,49 +549,38 @@ mod tests {
         let connect_labels = LabelSet::from([("request_kind", "connect")]);
         let now = CurrentClock::now();
 
-        // This test checks the behavior of `recalculate_udp_avg_connect_processing_time_ns``
-        // when no connections have been recorded yet. The first call should
-        // handle division by zero gracefully and return an infinite average,
-        // which is the current behavior.
+        // This test checks the behavior of `recalculate_udp_avg_processing_time_ns`
+        // when no processed requests have been recorded yet. The first call should
+        // handle division by zero gracefully and set the first average to the 
+        // processing time of the first request.
 
-        // todo: the first average should be 2000ns, not infinity.
-        // This is because the first connection is not counted in the average
-        // calculation if the counter is increased after calculating the average.
-        // The problem is that we count requests when they are accepted, not
-        // when they are processed. And we calculate the average when the
-        // response is sent.
-
-        // First calculation: no connections recorded yet, should result in infinity
+        // First calculation: no processed requests recorded yet
         let processing_time_1 = Duration::from_nanos(2000);
         let avg_1 = repo
             .recalculate_udp_avg_processing_time_ns(processing_time_1, &connect_labels, now)
             .await;
 
+        // The first average should be the first processing time since processed_requests_total is 0
+        // When processed_requests_total == 0.0, new_avg = req_processing_time
         assert!(
             (avg_1 - 2000.0).abs() < f64::EPSILON,
             "First calculation should be 2000, but got {avg_1}"
         );
 
-        // Now add one connection and try again
-        let ipv4_labels = LabelSet::from([("server_binding_address_ip_family", "inet"), ("request_kind", "connect")]);
-        repo.increase_counter(&metric_name!(UDP_TRACKER_SERVER_REQUESTS_ACCEPTED_TOTAL), &ipv4_labels, now)
-            .await
-            .unwrap();
-
-        // Second calculation: 1 connection
+        // Second calculation: now we have one processed request (incremented during first call)
         let processing_time_2 = Duration::from_nanos(3000);
-        let connect_labels = LabelSet::from([("request_kind", "connect")]);
         let avg_2 = repo
             .recalculate_udp_avg_processing_time_ns(processing_time_2, &connect_labels, now)
             .await;
 
-        // There is one connection, so the average should be:
-        // 2000 + (3000 - 2000) / 1 = 2000 + 1000 = 3000
-        // This is because one connection is not counted yet in the average calculation,
-        // so the average is simply the processing time of the second connection.
+        // Moving average calculation: previous_avg + (new_value - previous_avg) / processed_requests_total
+        // After first call: processed_requests_total = 1, avg = 2000
+        // During second call: processed_requests_total incremented to 2
+        // new_avg = 2000 + (3000 - 2000) / 2 = 2000 + 500 = 2500
+        let expected_avg_2 = 2000.0 + (3000.0 - 2000.0) / 2.0;
         assert!(
-            (avg_2 - 3000.0).abs() < f64::EPSILON,
-            "Second calculation should be 3000ns, but got {avg_2}"
+            (avg_2 - expected_avg_2).abs() < f64::EPSILON,
+            "Second calculation should be {expected_avg_2}ns, but got {avg_2}"
         );
     }
 }