feat: CachedRedisStorage will now default to flushing a batch ASAP to…

limitador/src/storage/redis/mod.rs

@@ -7,7 +7,7 @@ mod redis_cached;
 mod redis_sync;
 mod scripts;
 
-pub const DEFAULT_FLUSHING_PERIOD_SEC: u64 = 1;
+pub const DEFAULT_FLUSHING_PERIOD_SEC: u64 = 0;
 pub const DEFAULT_MAX_CACHED_COUNTERS: usize = 10000;
 pub const DEFAULT_MAX_TTL_CACHED_COUNTERS_SEC: u64 = 5;
 pub const DEFAULT_TTL_RATIO_CACHED_COUNTERS: u64 = 10;

limitador/src/storage/redis/redis_cached.rs

@@ -18,6 +18,7 @@ use std::str::FromStr;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex};
 use std::time::{Duration, Instant, SystemTime};
+use tokio::sync::Notify;
 use tracing::{debug_span, error, warn, Instrument};
 
 // This is just a first version.
@@ -44,6 +45,7 @@ pub struct CachedRedisStorage {
  async_redis_storage: AsyncRedisStorage,
  redis_conn_manager: ConnectionManager,
  partitioned: Arc<AtomicBool>,
+ pub flush_notify: Arc<Notify>,
 }
 
 #[async_trait]
@@ -154,10 +156,10 @@ impl AsyncCounterStorage for CachedRedisStorage {
  self.cached_counters.increase_by(counter, delta);
  }
 
- // Batch or update depending on configuration
  let mut batcher = self.batcher_counter_updates.lock().unwrap();
  let now = SystemTime::now();
  for counter in counters.iter() {
+ // Update an existing batch entry or add a new batch entry
  match batcher.get_mut(counter) {
  Some(val) => {
  val.update(delta, counter.seconds(), now);
@@ -174,6 +176,9 @@ impl AsyncCounterStorage for CachedRedisStorage {
  }
  }
 
+ // ask the flusher to flush the batch
+ self.flush_notify.notify_one();
+
  Ok(Authorization::Ok)
  }
 
@@ -240,15 +245,31 @@ impl CachedRedisStorage {
  let batcher: Arc<Mutex<HashMap<Counter, AtomicExpiringValue>>> =
  Arc::new(Mutex::new(Default::default()));
 
+ let flush_notify = Arc::new(Notify::new());
+
  {
  let storage = async_redis_storage.clone();
  let counters_cache_clone = counters_cache.clone();
  let conn = redis_conn_manager.clone();
  let p = Arc::clone(&partitioned);
  let batcher_flusher = batcher.clone();
- let mut interval = tokio::time::interval(flushing_period);
+ let flush_notify = flush_notify.clone();
+
  tokio::spawn(async move {
  loop {
+ // Wait for a new flush request,
+ flush_notify.notified().await;
+
+ if flushing_period != Duration::ZERO {
+ // Set the flushing_period to reduce the load on Redis the downside to
+ // setting it the cached counters will not be as accurate as when it's zero.
+ tokio::time::sleep(flushing_period).await
+ }
+
+ // even if flushing_period is zero, the next batch/ will be growing while
+ // current batch is being executed against Redis. When under load, the flush
+ // frequency will proportional to batch execution latency.
+
  flush_batcher_and_update_counters(
  conn.clone(),
  batcher_flusher.clone(),
@@ -257,7 +278,6 @@ impl CachedRedisStorage {
  p.clone(),
  )
  .await;
- interval.tick().await;
  }
  });
  }
@@ -268,6 +288,7 @@ impl CachedRedisStorage {
  redis_conn_manager,
  async_redis_storage,
  partitioned,
+ flush_notify,
  })
  }