diff --git a/api-server/Cargo.lock b/api-server/Cargo.lock index 325c605..4d3d164 100644 --- a/api-server/Cargo.lock +++ b/api-server/Cargo.lock @@ -91,6 +91,7 @@ dependencies = [ "opentelemetry-otlp", "opentelemetry-semantic-conventions", "opentelemetry_sdk", + "r2d2", "rand 0.8.6", "redis", "regex", @@ -115,13 +116,10 @@ dependencies = [ ] [[package]] -name = "arc-swap" -version = "1.9.2" +name = "arcstr" +version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c049c0be4daef0b145cb3555416b3b8ef5b7888a38aea1a3a155801fe7b0810b" -dependencies = [ - "rustversion", -] +checksum = "03918c3dbd7701a85c6b9887732e2921175f26c350b4563841d0958c21d57e6d" [[package]] name = "ascii_utils" @@ -688,11 +686,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ba5a308b75df32fe02788e748662718f03fde005016435c444eea572398219fd" dependencies = [ "bytes", - "futures-core", "memchr", - "pin-project-lite", - "tokio", - "tokio-util", ] [[package]] @@ -1095,26 +1089,6 @@ dependencies = [ "windows-sys 0.61.2", ] -[[package]] -name = "event-listener" -version = "5.4.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5a23add41df1562121a9393cb065eab5146a1242410f23a644851e90cfd669d2" -dependencies = [ - "parking", - "pin-project-lite", -] - -[[package]] -name = "event-listener-strategy" -version = "0.5.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8be9f3dfaaffdae2972880079a491a1a8bb7cbed0b8dd7a347f668b4150a3b93" -dependencies = [ - "event-listener", - "pin-project-lite", -] - [[package]] name = "fast_chemail" version = "0.9.6" @@ -1420,6 +1394,15 @@ dependencies = [ "digest", ] +[[package]] +name = "home" +version = "0.5.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cc627f471c528ff0c4a49e1d5e60450c8f6461dd6d10ba9dcd3a61d3dff7728d" +dependencies = [ + "windows-sys 0.61.2", +] + [[package]] name = "http" version = "1.4.2" @@ -2218,6 +2201,16 @@ version = "2.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f38d5652c16fde515bb1ecef450ab0f6a219d619a7274976324d5e377f7dceba" +[[package]] +name = "parking_lot" +version = "0.12.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93857453250e3077bd71ff98b6a65ea6621a19bb0f559a85248955ac12c45a1a" +dependencies = [ + "lock_api", + "parking_lot_core", +] + [[package]] name = "parking_lot_core" version = "0.9.12" @@ -2492,6 +2485,19 @@ dependencies = [ "syn 2.0.118", ] +[[package]] +name = "prost-derive" +version = "0.14.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf" +dependencies = [ + "anyhow", + "itertools", + "proc-macro2", + "quote", + "syn 2.0.118", +] + [[package]] name = "prost-types" version = "0.14.4" @@ -2592,6 +2598,17 @@ version = "6.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" +[[package]] +name = "r2d2" +version = "0.8.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "51de85fb3fb6524929c8a2eb85e6b6d363de4e8c48f9e2c2eac4944abc181c93" +dependencies = [ + "log", + "parking_lot", + "scheduled-thread-pool", +] + [[package]] name = "rand" version = "0.8.6" @@ -2683,52 +2700,19 @@ version = "1.6.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e37a4ca5c6ca42aa3e6df2fd32b987a65d32a4c2159a6f3fe0fd1df306a2658f" dependencies = [ - "arc-swap", "arcstr", - "async-lock", - "backon", - "bytes", - "cfg-if", "combine", - "futures-channel", - "futures-util", "itoa", "num-bigint 0.5.1", "percent-encoding", - "pin-project-lite", + "r2d2", "ryu", "sha1_smol", "socket2 0.6.4", - "tokio", - "tokio-util", "url", "xxhash-rust", ] -[[package]] -name = "redis" -version = "0.27.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "09d8f99a4090c89cc489a94833c901ead69bfbf3877b4867d5482e321ee875bc" -dependencies = [ - "arc-swap", - "async-trait", - "bytes", - "combine", - "futures-util", - "itertools 0.13.0", - "itoa", - "num-bigint", - "percent-encoding", - "pin-project-lite", - "ryu", - "sha1_smol", - "socket2 0.5.10", - "tokio", - "tokio-util", - "url", -] - [[package]] name = "redox_syscall" version = "0.5.18" @@ -2938,6 +2922,15 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "scheduled-thread-pool" +version = "0.2.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3cbc66816425a074528352f5789333ecff06ca41b36b0b0efdfbb29edc391a19" +dependencies = [ + "parking_lot", +] + [[package]] name = "schemars" version = "0.9.0" diff --git a/api-server/Cargo.toml b/api-server/Cargo.toml index 5059c32..5c989f9 100644 --- a/api-server/Cargo.toml +++ b/api-server/Cargo.toml @@ -53,4 +53,12 @@ regex = "1" base64 = "0.22" flate2 = "1.0" brotli = "7.0" -redis = { version = "0.27", features = ["tokio-comp"] } +redis = { version = "1", features = ["r2d2"] } +r2d2 = "0.8" + +[dev-dependencies] +testcontainers = { version = "0.27", features = ["blocking"] } +testcontainers-modules = { version = "0.15", features = ["redis"] } + +[features] +redis-integration-tests = [] diff --git a/api-server/src/cache.rs b/api-server/src/cache.rs index 05969fc..697afb4 100644 --- a/api-server/src/cache.rs +++ b/api-server/src/cache.rs @@ -1,12 +1,26 @@ /// #316: Redis-based caching layer for IP and Swap queries. /// -/// Uses an in-process DashMap as a TTL cache when Redis is unavailable, -/// falling back gracefully so the server always starts without Redis. -use std::sync::Arc; +/// Backed by Redis when `REDIS_URL` is configured and reachable, so that +/// invalidations (`invalidate`, `invalidate_prefix`, `invalidate_pattern`) +/// are visible to every `api-server` instance sharing that Redis, not just +/// the instance that performed the write. +/// +/// Falls back to an in-process `DashMap` TTL cache — gracefully, so the +/// server always starts and serves correct (if not shared) data — whenever +/// Redis is not configured, not reachable at startup, or becomes +/// unreachable while running. That degraded state is *not* silent: it is +/// tracked in [`is_degraded`], logged on each transition, and exposed via +/// the `cache_backend_degraded_transitions_total` counter. A background +/// thread pings Redis every [`HEALTH_CHECK_INTERVAL`] and flips the cache +/// back to shared mode automatically once Redis is reachable again. +use std::sync::atomic::{AtomicBool, Ordering}; use std::time::{Duration, Instant}; use dashmap::DashMap; +use metrics::{counter, describe_counter}; use once_cell::sync::Lazy; +use r2d2::Pool; +use redis::Commands; use serde::{de::DeserializeOwned, Serialize}; const DEFAULT_TTL_SECS: u64 = 30; @@ -14,6 +28,12 @@ const IP_TTL_SECS: u64 = 60; const SWAP_TTL_SECS: u64 = 30; const REPUTATION_TTL_SECS: u64 = 300; +/// How often the background thread pings Redis to detect recovery from a +/// degraded state. Cache operations do not retry Redis on every call while +/// degraded — they defer to this thread — so this interval is also the +/// worst-case time to resume shared caching after Redis comes back. +const HEALTH_CHECK_INTERVAL: Duration = Duration::from_secs(10); + struct Entry { value: String, expires_at: Instant, @@ -21,6 +41,137 @@ struct Entry { static STORE: Lazy> = Lazy::new(DashMap::new); +/// `true` when the cache is serving from the in-process `DashMap` fallback +/// instead of the shared Redis store — either because `REDIS_URL` isn't +/// configured, or because Redis is currently unreachable. Starts `true` and +/// flips to `false` once a real connection is confirmed. +static DEGRADED: AtomicBool = AtomicBool::new(true); + +/// Connection pool to the shared Redis store, built once from `REDIS_URL`. +/// `None` when `REDIS_URL` is unset or invalid, in which case the cache +/// runs in-process only and no health-check thread is started. +static REDIS_POOL: Lazy>> = Lazy::new(|| { + describe_counter!( + "cache_backend_degraded_transitions_total", + "Transitions of the cache between shared Redis mode and degraded in-process mode" + ); + + let url = match std::env::var("REDIS_URL") { + Ok(url) => url, + Err(_) => { + tracing::warn!("cache: REDIS_URL not set, running in-process memory cache only"); + return None; + } + }; + + let client = match redis::Client::open(url) { + Ok(client) => client, + Err(err) => { + tracing::warn!(error = %err, "cache: invalid REDIS_URL, running in-process memory cache only"); + return None; + } + }; + + // `build_unchecked` never fails and never blocks on a live connection — + // the server must always start even if Redis is down at boot. + let pool = Pool::builder() + .max_size(16) + .connection_timeout(Duration::from_millis(300)) + .build_unchecked(client); + + // Establish real reachability synchronously (bounded by the 300ms + // connection timeout above) so that the very first cache operation + // after startup sees an accurate `DEGRADED` state instead of racing the + // background health-check thread's first tick. + match ping(&pool) { + Ok(()) => mark_healthy(), + Err(reason) => mark_degraded(&reason), + } + + spawn_health_check(pool.clone()); + Some(pool) +}); + +fn ping(pool: &Pool) -> Result<(), String> { + pool.get() + .map_err(|err| err.to_string()) + .and_then(|mut conn| { + redis::cmd("PING") + .query::(&mut *conn) + .map(|_| ()) + .map_err(|err| err.to_string()) + }) +} + +fn spawn_health_check(pool: Pool) { + let spawned = std::thread::Builder::new() + .name("cache-redis-health".to_string()) + .spawn(move || loop { + std::thread::sleep(HEALTH_CHECK_INTERVAL); + match ping(&pool) { + Ok(()) => mark_healthy(), + Err(reason) => mark_degraded(&reason), + } + }); + + if let Err(err) = spawned { + tracing::error!(error = %err, "cache: failed to spawn Redis health-check thread; degraded state will only clear on the next successful cache operation"); + } +} + +/// Record a Redis failure. Logs and increments the transition counter only +/// on the edge (healthy -> degraded), not on every failed operation. +fn mark_degraded(reason: &str) { + let was_degraded = DEGRADED.swap(true, Ordering::SeqCst); + if !was_degraded { + tracing::warn!( + reason, + "cache: Redis unreachable, falling back to in-process memory cache (degraded mode)" + ); + counter!( + "cache_backend_degraded_transitions_total", + "direction" => "to_degraded", + ) + .increment(1); + } +} + +/// Record a successful Redis health check. Logs and increments the +/// transition counter only on the edge (degraded -> healthy). +fn mark_healthy() { + let was_degraded = DEGRADED.swap(false, Ordering::SeqCst); + if was_degraded { + tracing::info!("cache: Redis connection restored, resuming shared cache mode"); + counter!( + "cache_backend_degraded_transitions_total", + "direction" => "to_healthy", + ) + .increment(1); + } +} + +/// Whether the cache is currently serving from the in-process fallback +/// rather than shared Redis. +pub fn is_degraded() -> bool { + DEGRADED.load(Ordering::SeqCst) +} + +/// Get a pooled Redis connection, marking the cache degraded on failure. +/// Returns `None` when Redis isn't configured or is currently unreachable. +fn redis_conn() -> Option> { + let pool = REDIS_POOL.as_ref()?; + if is_degraded() { + return None; + } + match pool.get() { + Ok(conn) => Some(conn), + Err(err) => { + mark_degraded(&err.to_string()); + None + } + } +} + // ── Cache Configuration ─────────────────────────────────────────────────────── /// Configure cache TTL for different data types. @@ -44,7 +195,8 @@ impl Default for CacheConfig { } /// Global cache configuration. -static CONFIG: Lazy> = Lazy::new(|| Arc::new(CacheConfig::default())); +static CONFIG: Lazy> = + Lazy::new(|| std::sync::Arc::new(CacheConfig::default())); /// Initialize cache with custom configuration. /// Note: has no effect after the cache has been first accessed (Lazy is already initialized). @@ -62,19 +214,37 @@ pub fn set(key: &str, value: &T) { /// Write a value into the cache under `key` with custom TTL. pub fn set_with_ttl(key: &str, value: &T, ttl_secs: u64) { - if let Ok(json) = serde_json::to_string(value) { - STORE.insert( - key.to_string(), - Entry { - value: json, - expires_at: Instant::now() + Duration::from_secs(ttl_secs), - }, - ); + let Ok(json) = serde_json::to_string(value) else { + return; + }; + + if let Some(mut conn) = redis_conn() { + let result: redis::RedisResult<()> = conn.set_ex(key, &json, ttl_secs.max(1)); + match result { + Ok(()) => return, + Err(err) => mark_degraded(&err.to_string()), + } } + + STORE.insert( + key.to_string(), + Entry { + value: json, + expires_at: Instant::now() + Duration::from_secs(ttl_secs), + }, + ); } /// Read a cached value. Returns `None` on miss or expiry. pub fn get(key: &str) -> Option { + if let Some(mut conn) = redis_conn() { + match conn.get::<_, Option>(key) { + Ok(Some(json)) => return serde_json::from_str(&json).ok(), + Ok(None) => return None, + Err(err) => mark_degraded(&err.to_string()), + } + } + let entry = STORE.get(key)?; if entry.expires_at < Instant::now() { drop(entry); @@ -86,6 +256,13 @@ pub fn get(key: &str) -> Option { /// Check if a key exists and is not expired. pub fn exists(key: &str) -> bool { + if let Some(mut conn) = redis_conn() { + match conn.exists::<_, bool>(key) { + Ok(exists) => return exists, + Err(err) => mark_degraded(&err.to_string()), + } + } + match STORE.get(key) { Some(entry) => entry.expires_at >= Instant::now(), None => false, @@ -94,6 +271,14 @@ pub fn exists(key: &str) -> bool { /// Get TTL remaining for a key in seconds. Returns None if key doesn't exist or is expired. pub fn ttl_remaining(key: &str) -> Option { + if let Some(mut conn) = redis_conn() { + match conn.ttl::<_, i64>(key) { + Ok(ttl) if ttl >= 0 => return Some(ttl as u64), + Ok(_) => return None, + Err(err) => mark_degraded(&err.to_string()), + } + } + let entry = STORE.get(key)?; if entry.expires_at < Instant::now() { drop(entry); @@ -105,34 +290,86 @@ pub fn ttl_remaining(key: &str) -> Option { } /// Invalidate a single cache key. +/// +/// Removes the key from Redis (visible to every instance sharing it) and +/// from the local fallback store, so a flip between backends around the +/// time of the call can never leave a stale copy behind. pub fn invalidate(key: &str) { + if let Some(mut conn) = redis_conn() { + let result: redis::RedisResult<()> = conn.del(key); + if let Err(err) = result { + mark_degraded(&err.to_string()); + } + } STORE.remove(key); } /// Invalidate all keys that start with `prefix`. pub fn invalidate_prefix(prefix: &str) { + if let Some(mut conn) = redis_conn() { + match redis_delete_matching(&mut conn, &format!("{prefix}*")) { + Ok(()) => {} + Err(err) => mark_degraded(&err.to_string()), + } + } STORE.retain(|k, _| !k.starts_with(prefix)); } /// Invalidate all keys matching a pattern (supports * wildcards). pub fn invalidate_pattern(pattern: &str) { - if pattern.contains('*') { - let regex_pattern = pattern.replace('*', ".*"); - if let Ok(regex) = regex::Regex::new(®ex_pattern) { - STORE.retain(|k, _| !regex.is_match(k)); - } - } else { + if !pattern.contains('*') { invalidate_prefix(pattern); + return; } + + if let Some(mut conn) = redis_conn() { + // Redis glob patterns already use `*`/`?`/`[...]`, so the pattern is + // passed through to SCAN MATCH as-is — no regex translation needed. + match redis_delete_matching(&mut conn, pattern) { + Ok(()) => {} + Err(err) => mark_degraded(&err.to_string()), + } + } + + let regex_pattern = pattern.replace('*', ".*"); + if let Ok(regex) = regex::Regex::new(®ex_pattern) { + STORE.retain(|k, _| !regex.is_match(k)); + } +} + +/// Scan for keys matching a Redis glob `pattern` and delete them. +fn redis_delete_matching(conn: &mut redis::Connection, pattern: &str) -> redis::RedisResult<()> { + let keys: Vec = conn + .scan_match::<_, String>(pattern)? + .collect::, redis::RedisError>>()?; + if !keys.is_empty() { + let _: () = conn.del(keys)?; + } + Ok(()) } /// Clear all cache entries. +/// +/// When Redis-backed, this flushes the connected Redis database — deployments +/// that need `clear()` scoped strictly to this cache's keys should point +/// `REDIS_URL` at a Redis instance/logical DB dedicated to it. pub fn clear() { + if let Some(mut conn) = redis_conn() { + let result: redis::RedisResult<()> = redis::cmd("FLUSHDB").query(&mut *conn); + if let Err(err) = result { + mark_degraded(&err.to_string()); + } + } STORE.clear(); } /// Get cache statistics. pub fn stats() -> CacheStats { + if let Some(mut conn) = redis_conn() { + if let Ok(total_entries) = redis::cmd("DBSIZE").query::(&mut *conn) { + return CacheStats { total_entries }; + } + } CacheStats { total_entries: STORE.len(), } @@ -373,4 +610,21 @@ mod tests { assert!(!exists("ip:1")); assert!(!exists("ip:list:owner:10:0")); } + + #[test] + fn test_degraded_by_default_without_redis_url() { + // In the test binary REDIS_URL is unset, so the cache runs + // in-process only and reports itself as degraded — not silently. + assert!(is_degraded()); + } } + +// Redis integration tests (real Redis via testcontainers, gated behind the +// `redis-integration-tests` feature) live in `api-server/tests/`, not here. +// +// They must run as separate integration-test binaries — each `tests/*.rs` +// file gets its own process — because `REDIS_POOL` above is a `Lazy` that +// reads `REDIS_URL` exactly once per process, on first cache access. A +// `#[cfg(test)]` module in this file would share a process (and therefore +// the same already-initialized `REDIS_POOL`) with the in-process unit tests +// above, which rely on `REDIS_URL` staying unset. diff --git a/api-server/tests/cache_redis_cross_instance.rs b/api-server/tests/cache_redis_cross_instance.rs new file mode 100644 index 0000000..f2d3247 --- /dev/null +++ b/api-server/tests/cache_redis_cross_instance.rs @@ -0,0 +1,53 @@ +#![cfg(feature = "redis-integration-tests")] +//! Integration test for #786: cross-instance cache invalidation via Redis. +//! +//! Requires Docker and is gated behind the `redis-integration-tests` feature +//! so a plain `cargo test` never needs a Docker daemon: +//! +//! ```sh +//! cargo test --features redis-integration-tests --test cache_redis_cross_instance +//! ``` + +use api_server::cache; +use serde::{Deserialize, Serialize}; +use testcontainers::runners::SyncRunner; +use testcontainers_modules::redis::Redis; + +#[derive(Debug, Serialize, Deserialize, PartialEq)] +struct Dummy { + val: u64, +} + +/// `cache`'s Redis pool is a `Lazy` read from `REDIS_URL` on first access, so +/// the env var must be set before any `cache::` call in this process — this +/// integration binary contains only Redis-backed tests, so that's safe here. +#[test] +fn cross_instance_invalidation_is_visible_immediately() { + let container = Redis::default() + .start() + .expect("failed to start Redis container"); + let host = container.get_host().expect("failed to get container host"); + let port = container + .get_host_port_ipv4(6379) + .expect("failed to get mapped Redis port"); + std::env::set_var("REDIS_URL", format!("redis://{host}:{port}")); + + let key = "test:cross_instance:1"; + let value = Dummy { val: 7 }; + + // "Instance A" writes the value. + cache::set(key, &value); + assert!( + !cache::is_degraded(), + "expected the cache to be Redis-backed against a live container" + ); + assert_eq!(cache::get::(key), Some(Dummy { val: 7 })); + + // "Instance B" invalidates it. Because both "instances" share the same + // Redis, this issues a real DEL against the shared store rather than a + // local-only removal, so the invalidation is visible everywhere. + cache::invalidate(key); + + // "Instance A" must no longer observe the stale value. + assert_eq!(cache::get::(key), None); +} diff --git a/api-server/tests/cache_redis_fallback.rs b/api-server/tests/cache_redis_fallback.rs new file mode 100644 index 0000000..aa251d5 --- /dev/null +++ b/api-server/tests/cache_redis_fallback.rs @@ -0,0 +1,33 @@ +#![cfg(feature = "redis-integration-tests")] +//! Integration test for #786: the documented fallback path when Redis is +//! configured but unreachable. +//! +//! Gated behind the `redis-integration-tests` feature; run with: +//! ```sh +//! cargo test --features redis-integration-tests --test cache_redis_fallback +//! ``` + +use api_server::cache; +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Serialize, Deserialize, PartialEq)] +struct Dummy { + val: u64, +} + +/// Kept in its own binary (separate process from +/// `cache_redis_cross_instance`) because `cache`'s Redis pool reads +/// `REDIS_URL` once per process on first access. +#[test] +fn fallback_serves_correct_data_when_redis_unreachable() { + // Nothing listens on this port, so every Redis attempt fails fast and + // the cache falls back to in-process memory rather than erroring. + std::env::set_var("REDIS_URL", "redis://127.0.0.1:1"); + + let key = "test:fallback:1"; + let value = Dummy { val: 3 }; + + cache::set(key, &value); + assert_eq!(cache::get::(key), Some(Dummy { val: 3 })); + assert!(cache::is_degraded()); +}