Merge pull request #283 from Kuadrant/redis-cached-integration-tests

integration tests for async redis with local cache

limitador/src/storage/redis/counters_cache.rs

@@ -63,33 +63,32 @@ impl CountersCache {
         redis_ttl_ms: i64,
         ttl_margin: Duration,
     ) {
-        let counter_val = Self::value_from_redis_val(redis_val, counter.max_value());
-        let counter_ttl = self.ttl_from_redis_ttl(redis_ttl_ms, counter.seconds(), counter_val);
+        let counter_val = redis_val.unwrap_or(0);
+        let counter_ttl = self.ttl_from_redis_ttl(
+            redis_ttl_ms,
+            counter.seconds(),
+            counter_val,
+            counter.max_value(),
+        );
         if let Some(ttl) = counter_ttl.checked_sub(ttl_margin) {
             if ttl > Duration::from_secs(0) {
                 self.cache.insert(counter, counter_val, ttl);
             }
         }
     }
 
-    pub fn decrease_by(&mut self, counter: &Counter, delta: i64) {
+    pub fn increase_by(&mut self, counter: &Counter, delta: i64) {
         if let Some(val) = self.cache.get_mut(counter) {
-            *val -= delta
+            *val += delta
         };
     }
 
-    fn value_from_redis_val(redis_val: Option<i64>, counter_max: i64) -> i64 {
-        match redis_val {
-            Some(val) => val,
-            None => counter_max,
-        }
-    }
-
     fn ttl_from_redis_ttl(
         &self,
         redis_ttl_ms: i64,
         counter_seconds: u64,
         counter_val: i64,
+        counter_max: i64,
     ) -> Duration {
         // Redis returns -2 when the key does not exist. Ref:
         // https://redis.io/commands/ttl
@@ -102,11 +101,11 @@ impl CountersCache {
             Duration::from_secs(counter_seconds)
         };
 
-        // If a counter is already at 0, we can cache it for as long as its TTL
+        // If a counter is already at counter_max, we can cache it for as long as its TTL
         // is in Redis. This does not depend on the requests received by other
         // instances of Limitador. No matter what they do, we know that the
         // counter is not going to recover its quota until it expires in Redis.
-        if counter_val <= 0 {
+        if counter_val >= counter_max {
             return counter_ttl;
         }
 
@@ -205,7 +204,7 @@ mod tests {
     }
 
     #[test]
-    fn insert_saves_max_value_when_redis_val_is_none() {
+    fn insert_saves_0_when_redis_val_is_none() {
         let max_val = 10;
         let mut values = HashMap::new();
         values.insert("app_id".to_string(), "1".to_string());
@@ -223,13 +222,13 @@ mod tests {
         let mut cache = CountersCacheBuilder::new().build();
         cache.insert(counter.clone(), None, 10, Duration::from_secs(0));
 
-        assert_eq!(cache.get(&counter).unwrap(), max_val);
+        assert_eq!(cache.get(&counter).unwrap(), 0);
     }
 
     #[test]
-    fn decrease_by() {
+    fn increase_by() {
         let current_val = 10;
-        let decrease_by = 8;
+        let increase_by = 8;
         let mut values = HashMap::new();
         values.insert("app_id".to_string(), "1".to_string());
         let counter = Counter::new(
@@ -250,8 +249,8 @@ mod tests {
             10,
             Duration::from_secs(0),
         );
-        cache.decrease_by(&counter, decrease_by);
+        cache.increase_by(&counter, increase_by);
 
-        assert_eq!(cache.get(&counter).unwrap(), current_val - decrease_by);
+        assert_eq!(cache.get(&counter).unwrap(), current_val + increase_by);
     }
 }

limitador/src/storage/redis/redis_cached.rs

@@ -83,7 +83,7 @@ impl AsyncCounterStorage for CachedRedisStorage {
             for counter in counters.iter_mut() {
                 match cached_counters.get(counter) {
                     Some(val) => {
-                        if first_limited.is_none() && val - delta < 0 {
+                        if first_limited.is_none() && val + delta > counter.max_value() {
                             let a = Authorization::Limited(
                                 counter.limit().name().map(|n| n.to_owned()),
                             );
@@ -93,7 +93,7 @@ impl AsyncCounterStorage for CachedRedisStorage {
                             first_limited = Some(a);
                         }
                         if load_counters {
-                            counter.set_remaining(val);
+                            counter.set_remaining(counter.max_value() - val - delta);
                             // todo: how do we get the ttl for this entry?
                             // counter.set_expires_in(Duration::from_secs(counter.seconds()));
                         }
@@ -128,15 +128,21 @@ impl AsyncCounterStorage for CachedRedisStorage {
                         counter_ttls_msecs[i],
                         ttl_margin,
                     );
-                    let remaining = counter_vals[i].unwrap_or(counter.max_value()) - delta;
+                    let remaining = counter.max_value() - counter_vals[i].unwrap_or(0) - delta;
                     if first_limited.is_none() && remaining < 0 {
                         first_limited = Some(Authorization::Limited(
                             counter.limit().name().map(|n| n.to_owned()),
                         ));
                     }
                     if load_counters {
                         counter.set_remaining(remaining);
-                        counter.set_expires_in(Duration::from_millis(counter_ttls_msecs[i] as u64));
+                        let counter_ttl = if counter_ttls_msecs[i] >= 0 {
+                            Duration::from_millis(counter_ttls_msecs[i] as u64)
+                        } else {
+                            Duration::from_secs(counter.max_value() as u64)
+                        };
+
+                        counter.set_expires_in(counter_ttl);
                     }
                 }
             }
@@ -150,7 +156,7 @@ impl AsyncCounterStorage for CachedRedisStorage {
         {
             let mut cached_counters = self.cached_counters.lock().unwrap();
             for counter in counters.iter() {
-                cached_counters.decrease_by(counter, delta);
+                cached_counters.increase_by(counter, delta);
             }
         }
 

limitador/tests/integration_tests.rs

@@ -24,7 +24,7 @@ macro_rules! test_with_all_storage_impls {
             #[cfg(feature = "redis_storage")]
             #[tokio::test]
             #[serial]
-            async fn [<$function _with_redis>]() {
+            async fn [<$function _with_sync_redis>]() {
                 let storage = RedisStorage::default();
                 storage.clear().unwrap();
                 let rate_limiter = RateLimiter::new_with_storage(
@@ -51,7 +51,23 @@ macro_rules! test_with_all_storage_impls {
                 let rate_limiter = AsyncRateLimiter::new_with_storage(
                     Box::new(storage)
                 );
-                AsyncRedisStorage::new("redis://127.0.0.1:6379").await.expect("We need a Redis running locally").clear().await.unwrap();
+                $function(&mut TestsLimiter::new_from_async_impl(rate_limiter)).await;
+            }
+
+            #[cfg(feature = "redis_storage")]
+            #[tokio::test]
+            #[serial]
+            async fn [<$function _with_async_redis_and_local_cache>]() {
+                let storage_builder = CachedRedisStorageBuilder::new("redis://127.0.0.1:6379").
+                    flushing_period(Some(Duration::from_millis(200))).
+                    max_ttl_cached_counters(Duration::from_secs(3600)).
+                    ttl_ratio_cached_counters(1).
+                    max_cached_counters(10000);
+                let storage = storage_builder.build().await.expect("We need a Redis running locally");
+                storage.clear().await.unwrap();
+                let rate_limiter = AsyncRateLimiter::new_with_storage(
+                    Box::new(storage)
+                );
                 $function(&mut TestsLimiter::new_from_async_impl(rate_limiter)).await;
             }
 
@@ -82,6 +98,7 @@ mod test {
     cfg_if::cfg_if! {
         if #[cfg(feature = "redis_storage")] {
             use limitador::storage::redis::AsyncRedisStorage;
+            use limitador::storage::redis::CachedRedisStorageBuilder;
             use limitador::storage::redis::RedisStorage;
 
             use limitador::AsyncRateLimiter;