Merge pull request #285 from Kuadrant/batch_update

Batch update

limitador/Cargo.toml

@@ -54,6 +54,7 @@ base64 = { version = "0.22", optional = true }
 [dev-dependencies]
 serial_test = "3.0"
 criterion = { version = "0.5.1", features = ["html_reports"] }
+redis-test = { version = "0.4.0", features = ["aio"] }
 paste = "1"
 rand = "0.8"
 tempfile = "3.5.0"

limitador/src/storage/keys.rs

@@ -36,7 +36,7 @@ pub fn prefix_for_namespace(namespace: &str) -> String {
 }
 
 pub fn counter_from_counter_key(key: &str, limit: &Limit) -> Counter {
-    let mut counter = partial_counter_from_counter_key(key, limit.namespace().as_ref());
+    let mut counter = partial_counter_from_counter_key(key);
     if !counter.update_to_limit(limit) {
         // this means some kind of data corruption _or_ most probably
         // an out of sync `impl PartialEq for Limit` vs `pub fn key_for_counter(counter: &Counter) -> String`
@@ -49,11 +49,24 @@ pub fn counter_from_counter_key(key: &str, limit: &Limit) -> Counter {
     counter
 }
 
-pub fn partial_counter_from_counter_key(key: &str, namespace: &str) -> Counter {
-    let offset = ",counter:".len();
-    let start_pos_counter = prefix_for_namespace(namespace).len() + offset;
-
-    let counter: Counter = serde_json::from_str(&key[start_pos_counter..]).unwrap();
+pub fn partial_counter_from_counter_key(key: &str) -> Counter {
+    let namespace_prefix = "namespace:";
+    let counter_prefix = ",counter:";
+
+    // Find the start position of the counter portion
+    let start_pos_namespace = key
+        .find(namespace_prefix)
+        .expect("Namespace not found in the key");
+    let start_pos_counter = key[start_pos_namespace..]
+        .find(counter_prefix)
+        .expect("Counter not found in the key")
+        + start_pos_namespace
+        + counter_prefix.len();
+
+    // Extract counter JSON substring and deserialize it
+    let counter_str = &key[start_pos_counter..];
+    let counter: Counter =
+        serde_json::from_str(counter_str).expect("Failed to deserialize counter JSON");
     counter
 }
 
@@ -87,7 +100,7 @@ mod tests {
         let limit = Limit::new(namespace, 1, 1, vec!["req.method == 'GET'"], vec!["app_id"]);
         let counter = Counter::new(limit.clone(), HashMap::default());
         let raw = key_for_counter(&counter);
-        assert_eq!(counter, partial_counter_from_counter_key(&raw, namespace));
+        assert_eq!(counter, partial_counter_from_counter_key(&raw));
         let prefix = prefix_for_namespace(namespace);
         assert_eq!(&raw[0..prefix.len()], &prefix);
     }

limitador/src/storage/redis/counters_cache.rs

@@ -121,9 +121,10 @@ impl CountersCache {
         );
         if let Some(ttl) = cache_ttl.checked_sub(ttl_margin) {
             if ttl > Duration::ZERO {
-                let value = CachedCounterValue::from(&counter, counter_val, cache_ttl);
-                let previous = self.cache.get_with(counter.clone(), || Arc::new(value));
-                if previous.expired_at(now) {
+                let previous = self.cache.get_with(counter.clone(), || {
+                    Arc::new(CachedCounterValue::from(&counter, counter_val, cache_ttl))
+                });
+                if previous.expired_at(now) || previous.value.value() < counter_val {
                     previous.set_from_authority(&counter, counter_val, cache_ttl);
                 }
                 return previous;