Merge pull request #317 from Kuadrant/absolute_ts

Use absolute timestamps when dealing with expiry from Redis

.github/workflows/rust.yml

@@ -38,7 +38,7 @@ jobs:
       - uses: actions/checkout@v4
       - uses: supercharge/[email protected]
         with:
-          redis-version: 5
+          redis-version: 7
       - uses: actions-rust-lang/setup-rust-toolchain@v1
       - uses: abelfodil/protoc-action@v1
         with:

limitador-server/src/config.rs

@@ -163,8 +163,6 @@ pub struct RedisStorageConfiguration {
 #[derive(PartialEq, Eq, Debug)]
 pub struct RedisStorageCacheConfiguration {
     pub flushing_period: i64,
-    pub max_ttl: u64,
-    pub ttl_ratio: u64,
     pub max_counters: usize,
     pub response_timeout: u64,
 }

limitador-server/src/main.rs

@@ -25,8 +25,7 @@ use limitador::storage::disk::DiskStorage;
 use limitador::storage::infinispan::{Consistency, InfinispanStorageBuilder};
 use limitador::storage::redis::{
     AsyncRedisStorage, CachedRedisStorage, CachedRedisStorageBuilder, DEFAULT_FLUSHING_PERIOD_SEC,
-    DEFAULT_MAX_CACHED_COUNTERS, DEFAULT_MAX_TTL_CACHED_COUNTERS_SEC, DEFAULT_RESPONSE_TIMEOUT_MS,
-    DEFAULT_TTL_RATIO_CACHED_COUNTERS,
+    DEFAULT_MAX_CACHED_COUNTERS, DEFAULT_RESPONSE_TIMEOUT_MS,
 };
 use limitador::storage::{AsyncCounterStorage, AsyncStorage, Storage};
 use limitador::{
@@ -134,8 +133,6 @@ impl Limiter {
 
         let cached_redis_storage = CachedRedisStorageBuilder::new(redis_url)
             .flushing_period(Duration::from_millis(cache_cfg.flushing_period as u64))
-            .max_ttl_cached_counters(Duration::from_millis(cache_cfg.max_ttl))
-            .ttl_ratio_cached_counters(cache_cfg.ttl_ratio)
             .max_cached_counters(cache_cfg.max_counters)
             .response_timeout(Duration::from_millis(cache_cfg.response_timeout));
 
@@ -591,30 +588,6 @@ fn create_config() -> (Configuration, &'static str) {
                 .about("Uses Redis to store counters, with an in-memory cache")
                 .display_order(4)
                 .arg(redis_url_arg)
-                .arg(
-                    Arg::new("TTL")
-                        .long("ttl")
-                        .action(ArgAction::Set)
-                        .value_parser(clap::value_parser!(u64))
-                        .default_value(
-                            config::env::REDIS_LOCAL_CACHE_MAX_TTL_CACHED_COUNTERS_MS
-                                .unwrap_or(leak(DEFAULT_MAX_TTL_CACHED_COUNTERS_SEC * 1000)),
-                        )
-                        .display_order(2)
-                        .help("TTL for cached counters in milliseconds"),
-                )
-                .arg(
-                    Arg::new("ratio")
-                        .long("ratio")
-                        .action(ArgAction::Set)
-                        .value_parser(clap::value_parser!(u64))
-                        .default_value(
-                            config::env::REDIS_LOCAL_CACHE_TTL_RATIO_CACHED_COUNTERS
-                                .unwrap_or(leak(DEFAULT_TTL_RATIO_CACHED_COUNTERS)),
-                        )
-                        .display_order(3)
-                        .help("Ratio to apply to the TTL from Redis on cached counters"),
-                )
                 .arg(
                     Arg::new("flush")
                         .long("flush-period")
@@ -748,8 +721,6 @@ fn create_config() -> (Configuration, &'static str) {
             url: sub.get_one::<String>("URL").unwrap().to_owned(),
             cache: Some(RedisStorageCacheConfiguration {
                 flushing_period: *sub.get_one("flush").unwrap(),
-                max_ttl: *sub.get_one("TTL").unwrap(),
-                ttl_ratio: *sub.get_one("ratio").unwrap(),
                 max_counters: *sub.get_one("max").unwrap(),
                 response_timeout: *sub.get_one("timeout").unwrap(),
             }),
@@ -832,16 +803,6 @@ fn storage_config_from_env() -> Result<StorageConfiguration, ()> {
                         .unwrap_or_else(|_| (DEFAULT_FLUSHING_PERIOD_SEC * 1000).to_string())
                         .parse()
                         .expect("Expected an i64"),
-                    max_ttl: env::var("REDIS_LOCAL_CACHE_MAX_TTL_CACHED_COUNTERS_MS")
-                        .unwrap_or_else(|_| {
-                            (DEFAULT_MAX_TTL_CACHED_COUNTERS_SEC * 1000).to_string()
-                        })
-                        .parse()
-                        .expect("Expected an u64"),
-                    ttl_ratio: env::var("REDIS_LOCAL_CACHE_TTL_RATIO_CACHED_COUNTERS")
-                        .unwrap_or_else(|_| DEFAULT_TTL_RATIO_CACHED_COUNTERS.to_string())
-                        .parse()
-                        .expect("Expected an u64"),
                     max_counters: DEFAULT_MAX_CACHED_COUNTERS,
                     response_timeout: DEFAULT_RESPONSE_TIMEOUT_MS,
                 })

limitador/src/storage/atomic_expiring_value.rs

@@ -28,8 +28,8 @@ impl AtomicExpiringValue {
     }
 
     #[allow(dead_code)]
-    pub fn add_and_set_expiry(&self, delta: i64, expiry: Duration) -> i64 {
-        self.expiry.update(expiry);
+    pub fn add_and_set_expiry(&self, delta: i64, expire_at: SystemTime) -> i64 {
+        self.expiry.update(expire_at);
         self.value.fetch_add(delta, Ordering::SeqCst) + delta
     }
 
@@ -44,12 +44,6 @@ impl AtomicExpiringValue {
     pub fn ttl(&self) -> Duration {
         self.expiry.duration()
     }
-
-    #[allow(dead_code)]
-    pub fn set(&self, value: i64, ttl: Duration) {
-        self.expiry.update(ttl);
-        self.value.store(value, Ordering::SeqCst);
-    }
 }
 
 #[derive(Debug)]
@@ -90,9 +84,9 @@ impl AtomicExpiryTime {
     }
 
     #[allow(dead_code)]
-    pub fn update(&self, ttl: Duration) {
+    pub fn update(&self, expiry: SystemTime) {
         self.expiry
-            .store(Self::since_epoch(SystemTime::now() + ttl), Ordering::SeqCst);
+            .store(Self::since_epoch(expiry), Ordering::SeqCst);
     }
 
     pub fn update_if_expired(&self, ttl: u64, when: SystemTime) -> bool {