Resilient cache

limitador-server/examples/limits.yaml

@@ -14,4 +14,4 @@
  conditions:
  - "req.method == 'POST'"
  variables:
- - user_id
+ - user_id

limitador-server/src/config.rs

@@ -166,6 +166,7 @@ pub struct RedisStorageCacheConfiguration {
  pub max_ttl: u64,
  pub ttl_ratio: u64,
  pub max_counters: usize,
+ pub response_timeout: u64,
 }
 
 #[derive(PartialEq, Eq, Debug)]

limitador-server/src/main.rs

@@ -23,7 +23,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_MAX_CACHED_COUNTERS, DEFAULT_MAX_TTL_CACHED_COUNTERS_SEC, DEFAULT_RESPONSE_TIMEOUT_MS,
  DEFAULT_TTL_RATIO_CACHED_COUNTERS,
 };
 use limitador::storage::{AsyncCounterStorage, AsyncStorage, Storage};
@@ -137,29 +137,17 @@ impl Limiter {
  ) -> CachedRedisStorage {
  // TODO: Not all the options are configurable via ENV. Add them as needed.
 
- let mut cached_redis_storage = CachedRedisStorageBuilder::new(redis_url);
+ 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));
 
- if cache_cfg.flushing_period < 0 {
- cached_redis_storage = cached_redis_storage.flushing_period(None)
- } else {
- cached_redis_storage = cached_redis_storage.flushing_period(Some(
- Duration::from_millis(cache_cfg.flushing_period as u64),
- ))
- }
-
- cached_redis_storage =
- cached_redis_storage.max_ttl_cached_counters(Duration::from_millis(cache_cfg.max_ttl));
-
- cached_redis_storage = cached_redis_storage.ttl_ratio_cached_counters(cache_cfg.ttl_ratio);
- cached_redis_storage = cached_redis_storage.max_cached_counters(cache_cfg.max_counters);
-
- match cached_redis_storage.build().await {
- Ok(storage) => storage,
- Err(err) => {
- eprintln!("Failed to connect to Redis at {redis_url}: {err}");
- process::exit(1)
- }
- }
+ cached_redis_storage.build().await.unwrap_or_else(|err| {
+ eprintln!("Failed to connect to Redis at {redis_url}: {err}");
+ process::exit(1)
+ })
  }
 
  #[cfg(feature = "infinispan")]
@@ -653,6 +641,15 @@ fn create_config() -> (Configuration, &'static str) {
  .default_value(leak(DEFAULT_MAX_CACHED_COUNTERS))
  .display_order(5)
  .help("Maximum amount of counters cached"),
+ )
+ .arg(
+ Arg::new("timeout")
+ .long("response-timeout")
+ .action(ArgAction::Set)
+ .value_parser(clap::value_parser!(u64))
+ .default_value(leak(DEFAULT_RESPONSE_TIMEOUT_MS))
+ .display_order(6)
+ .help("Timeout for Redis commands in milliseconds"),
  ),
  );
 
@@ -760,6 +757,7 @@ fn create_config() -> (Configuration, &'static str) {
  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(),
  }),
  }),
  #[cfg(feature = "infinispan")]
@@ -851,6 +849,7 @@ fn storage_config_from_env() -> Result<StorageConfiguration, ()> {
  .parse()
  .expect("Expected an u64"),
  max_counters: DEFAULT_MAX_CACHED_COUNTERS,
+ response_timeout: DEFAULT_RESPONSE_TIMEOUT_MS,
  })
  } else {
  None

limitador/Cargo.toml

@@ -38,7 +38,7 @@ tracing = "0.1.40"
 
 # Optional dependencies
 rocksdb = { version = "0.21.0", optional = true, features = ["multi-threaded-cf"] }
-redis = { version = "0.23.1", optional = true, features = [
+redis = { version = "0.25.1", optional = true, features = [
  "connection-manager",
  "tokio-comp",
  "tls-native-tls",

limitador/src/storage/disk/expiring_value.rs

@@ -98,6 +98,7 @@ impl From<TryFromSliceError> for StorageErr {
  fn from(_: TryFromSliceError) -> Self {
  Self {
  msg: "Corrupted byte sequence while reading 8 bytes for 64-bit integer".to_owned(),
+ transient: false,
  }
  }
 }

limitador/src/storage/disk/mod.rs

@@ -1,4 +1,5 @@
 use crate::storage::StorageErr;
+use rocksdb::ErrorKind;
 
 mod expiring_value;
 mod rocksdb_storage;
@@ -9,6 +10,7 @@ impl From<rocksdb::Error> for StorageErr {
  fn from(error: rocksdb::Error) -> Self {
  Self {
  msg: format!("Underlying storage error: {error}"),
+ transient: error.kind() == ErrorKind::TimedOut || error.kind() == ErrorKind::TryAgain,
  }
  }
 }

limitador/src/storage/infinispan/dist_lock.rs

@@ -49,6 +49,7 @@ pub async fn lock(
  if retries >= RETRIES {
  return Err(StorageErr {
  msg: "can't acquire lock".into(),
+ transient: false,
  });
  }
 

limitador/src/storage/infinispan/mod.rs

@@ -12,13 +12,19 @@ pub use infinispan_storage::InfinispanStorageBuilder;
 
 impl From<reqwest::Error> for StorageErr {
  fn from(e: reqwest::Error) -> Self {
- Self { msg: e.to_string() }
+ Self {
+ msg: e.to_string(),
+ transient: false,
+ }
  }
 }
 
 impl From<InfinispanError> for StorageErr {
  fn from(e: InfinispanError) -> Self {
- Self { msg: e.to_string() }
+ Self {
+ msg: e.to_string(),
+ transient: false,
+ }
  }
 }
 

limitador/src/storage/mod.rs

@@ -299,10 +299,15 @@ pub trait AsyncCounterStorage: Sync + Send {
 #[error("error while accessing the limits storage: {msg}")]
 pub struct StorageErr {
  msg: String,
+ transient: bool,
 }
 
 impl StorageErr {
  pub fn msg(&self) -> &str {
  &self.msg
  }
+
+ pub fn is_transient(&self) -> bool {
+ self.transient
+ }
 }

limitador/src/storage/redis/counters_cache.rs

@@ -8,7 +8,7 @@ use ttl_cache::TtlCache;
 
 pub struct CountersCache {
  max_ttl_cached_counters: Duration,
- ttl_ratio_cached_counters: u64,
+ pub ttl_ratio_cached_counters: u64,
  cache: TtlCache<Counter, i64>,
 }
 
@@ -204,7 +204,7 @@ mod tests {
  }
 
  #[test]
- fn insert_saves_0_when_redis_val_is_none() {
+ fn insert_saves_zero_when_redis_val_is_none() {
  let max_val = 10;
  let mut values = HashMap::new();
  values.insert("app_id".to_string(), "1".to_string());

limitador/src/storage/redis/mod.rs

@@ -11,6 +11,7 @@ pub const DEFAULT_FLUSHING_PERIOD_SEC: u64 = 1;
 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;
+pub const DEFAULT_RESPONSE_TIMEOUT_MS: u64 = 350;
 
 use crate::counter::Counter;
 use crate::storage::{Authorization, StorageErr};
@@ -21,13 +22,10 @@ pub use redis_sync::RedisStorage;
 
 impl From<RedisError> for StorageErr {
  fn from(e: RedisError) -> Self {
- Self { msg: e.to_string() }
- }
-}
-
-impl From<::r2d2::Error> for StorageErr {
- fn from(e: ::r2d2::Error) -> Self {
- Self { msg: e.to_string() }
+ Self {
+ msg: e.to_string(),
+ transient: e.is_timeout() || e.is_connection_dropped() || e.is_cluster_error(),
+ }
  }
 }
 

limitador/src/storage/redis/redis_async.rs

@@ -241,6 +241,14 @@ impl AsyncRedisStorage {
  Self { conn_manager }
  }
 
+ pub async fn is_alive(&self) -> bool {
+ self.conn_manager
+ .clone()
+ .incr::<&str, i32, u64>("LIMITADOR_LIVE_CHECK", 1)
+ .await
+ .is_ok()
+ }
+
  async fn delete_counters_associated_with_limit(&self, limit: &Limit) -> Result<(), StorageErr> {
  let mut con = self.conn_manager.clone();