/* This file is part of DarkFi (https://dark.fi) * * Copyright (C) 2020-2026 Dyne.org foundation * * This program is free software: you can redistribute it and/or modify * it under the terms of the GNU Affero General Public License as * published by the Free Software Foundation, either version 3 of the * License, or (at your option) any later version. * * This program is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the * GNU Affero General Public License for more details. * * You should have received a copy of the GNU Affero General Public License * along with this program. If not, see . */ use std::collections::VecDeque; use tracing::debug; use crate::util::time::NanoTimestamp; /// Struct representing metering configuration parameters. #[derive(Clone, Debug)] pub struct MeteringConfiguration { /// Defines the threshold after which rate limit kicks in. /// Set to 0 for no threshold. /// /// If we don't use raw count as our metric, it should be calculated /// by multiplying the median increase of the measured item with the /// "max" number of items we want before rate limit starts. /// For example, if we measure some item that increases our total /// measurement by ~5 and want to rate limit after about 10, this /// should be set as 50. pub threshold: u64, /// Sleep time for each unit over the threshold, in milliseconds. /// /// This is used to calculate sleep time when ratelimit is active. /// The computed sleep time when we are over the threshold will be: /// sleep_time = (total - threshold) * sleep_step pub sleep_step: u64, /// Parameter defining the expiration of each item, for time based /// decay, in nano seconds. Set to 0 for no expiration. pub expiry_time: NanoTimestamp, } impl MeteringConfiguration { /// Generate a new `MeteringConfiguration` for provided threshold, /// sleep step and expiration time (seconds). pub fn new(threshold: u64, sleep_step: u64, expiry_time: u128) -> Self { Self { threshold, sleep_step, expiry_time: NanoTimestamp::from_secs(expiry_time) } } } impl Default for MeteringConfiguration { fn default() -> Self { Self { threshold: 0, sleep_step: 0, expiry_time: NanoTimestamp(0) } } } /// Default `MeteringConfiguration` as a constant, /// so it can be used in trait macros. pub const DEFAULT_METERING_CONFIGURATION: MeteringConfiguration = MeteringConfiguration { threshold: 0, sleep_step: 0, expiry_time: NanoTimestamp(0) }; /// Struct to keep track of some sequential metered actions and compute /// rate limits. /// /// The queue uses a time based decay and prunes metering information /// after corresponding expiration time has passed. #[derive(Debug)] pub struct MeteringQueue { /// Metering configuration of the queue. config: MeteringConfiguration, /// Ring buffer keeping track of action execution timestamp and /// its metered value. queue: VecDeque<(NanoTimestamp, u64)>, } impl MeteringQueue { /// Generate a new `MeteringQueue` for provided `MeteringConfiguration`. pub fn new(config: MeteringConfiguration) -> Self { Self { config, queue: VecDeque::new() } } /// Prune expired metering information from the queue. pub fn clean(&mut self) { // Check if expiration has been set if self.config.expiry_time.0 == 0 { return } // Iterate the queue to cleanup expired elements while let Some((ts, _)) = self.queue.front() { // This is an edge case where system reports a future timestamp // therefore elapsed computation fails. let Ok(elapsed) = ts.elapsed() else { debug!(target: "net::metering::MeteringQueue::clean", "Timestamp [{ts}] is in future. Removing..."); let _ = self.queue.pop_front(); continue }; // Check if elapsed time is over the expiration limit if elapsed < self.config.expiry_time { break } // Remove element let _ = self.queue.pop_front(); } } /// Add new metering value to the queue, after /// prunning expired metering information. /// If no threshold has been set, the insert is /// ignored. pub fn push(&mut self, value: &u64) { // Check if threshold has been set if self.config.threshold == 0 { return } // Prune expired elements self.clean(); // Push the new value self.queue.push_back((NanoTimestamp::current_time(), *value)); } /// Compute the current metered values total. pub fn total(&self) -> u64 { let mut total = 0; for (_, value) in &self.queue { total += value; } total } /// Compute sleep time for current metered values total, based on /// the metering configuration. /// /// The sleep time increases linearly, based on configuration sleep /// step. For example, in a raw count metering model, if we set the /// configuration with threshold = 6 and sleep_step = 250, when /// total = 10, returned sleep time will be 1000 ms. /// /// Sleep times table for the above example: /// /// | Total | Sleep Time (ms) | /// |-------|-----------------| /// | 0 | 0 | /// | 4 | 0 | /// | 6 | 0 | /// | 7 | 250 | /// | 8 | 500 | /// | 9 | 750 | /// | 10 | 1000 | /// | 14 | 2000 | /// | 18 | 3000 | pub fn sleep_time(&self) -> Option { // Check if threshold has been set if self.config.threshold == 0 { return None } // Check if we are over the threshold let total = self.total(); if total < self.config.threshold { return None } // Compute the actual sleep time Some((total - self.config.threshold) * self.config.sleep_step) } } #[test] fn test_net_metering_queue_default() { let mut queue = MeteringQueue::new(MeteringConfiguration::default()); for _ in 0..100 { queue.push(&1); assert!(queue.queue.is_empty()); assert_eq!(queue.total(), 0); assert!(queue.sleep_time().is_none()); } } #[test] fn test_net_metering_queue_raw_count() { let threshold = 6; let sleep_step = 250; let metering_configuration = MeteringConfiguration::new(threshold, sleep_step, 0); let mut queue = MeteringQueue::new(metering_configuration); for i in 1..threshold { queue.push(&1); assert_eq!(queue.total(), i); assert!(queue.sleep_time().is_none()); } for i in threshold..100 { queue.push(&1); assert_eq!(queue.total(), i); assert_eq!(queue.sleep_time(), Some((i - threshold) * sleep_step)); } } #[test] fn test_net_metering_queue_sleep_time() { let metered_value_median = 5; let threshold_items = 10; let threshold = metered_value_median * threshold_items; let sleep_step = 50; let metering_configuration = MeteringConfiguration::new(threshold, sleep_step, 0); let mut queue = MeteringQueue::new(metering_configuration); for i in 1..threshold_items { queue.push(&metered_value_median); assert_eq!(queue.total(), (i * metered_value_median)); assert!(queue.sleep_time().is_none()); } for i in threshold_items..100 { queue.push(&metered_value_median); let expected_total = i * metered_value_median; assert_eq!(queue.total(), expected_total); assert_eq!(queue.sleep_time(), Some((expected_total - threshold) * sleep_step)); } }