From 7a12f0e9c8f377c49dfbf391926685d0ab196280 Mon Sep 17 00:00:00 2001 From: asepharyana Date: Mon, 31 Aug 2026 20:11:04 +0700 Subject: [PATCH] fix(cache): self-heal Redis connection + don't 500 on cache write failure MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The scraper cached a single RedisCache (one multiplexed connection) in a OnceCell forever. When that one connection broke (Redis restart, idle timeout, network blip), every cache op failed with 'cache io: broken pipe', and because Cache::get_or_set propagated the post-compute write error, EVERY API (anime, anime2, komik) returned 500 until a process restart. Fix: - Build a fresh RedisCache from a freshly checked-out deadpool connection per call, so a broken connection self-heals without a process restart (deadpool recycles/drops dead connections and reconnects on checkout). - Make Cache::get_or_set treat a cache-write failure as non-fatal: return the freshly computed value (cache is best-effort), so a transient Redis outage degrades to cache-less instead of 500. Pre-existing clippy warnings (repositories/parsers) untouched — out of scope. --- src/infrastructure/cache/mytheclipse.rs | 49 ++++++++++++------------- src/infrastructure/cache/redis.rs | 23 +++++++++--- 2 files changed, 41 insertions(+), 31 deletions(-) diff --git a/src/infrastructure/cache/mytheclipse.rs b/src/infrastructure/cache/mytheclipse.rs index 8d4b4cb..65cb4ec 100644 --- a/src/infrastructure/cache/mytheclipse.rs +++ b/src/infrastructure/cache/mytheclipse.rs @@ -6,39 +6,38 @@ //! same `redis` crate version, a `deadpool_redis::Connection` can be converted //! directly into a `redis::aio::MultiplexedConnection` via `take()`. -use std::sync::LazyLock; - use mytheclipse_cache::{Cache, CacheError, RedisCache}; -use tokio::sync::OnceCell; use crate::infrastructure::cache::redis_pool::redis_pool; -/// Lazily-initialised shared `RedisCache` built from the deadpool pool. +/// Build a fresh `RedisCache` from a freshly checked-out deadpool connection +/// on each call. /// -/// The multiplexed connection is cheaply cloneable (Arc-backed), so the whole -/// process shares one logical connection while deadpool manages recycling. -static REDIS_CACHE: LazyLock> = LazyLock::new(OnceCell::new); - -/// Return a handle to the shared mytheclipse `RedisCache`, initialising it on -/// first use from the deadpool pool. -pub async fn redis_cache() -> Result<&'static RedisCache, CacheError> { - let cell = &*REDIS_CACHE; - cell.get_or_try_init(|| async { - let pool = redis_pool().map_err(CacheError::Io)?; - let conn = pool - .get() - .await - .map_err(|e| CacheError::Io(e.to_string()))?; - let mux = deadpool_redis::Connection::take(conn); - Ok(RedisCache::new(mux)) - }) - .await +/// Why fresh on every call (not a cached singleton): the previous design +/// initialised *one* `RedisCache` (one multiplexed connection) on first use and +/// kept it forever. If that single connection broke (Redis restart, idle +/// timeout, network blip), every cache operation failed with `broken pipe` +/// until the whole process restarted — and because `Cache::get_or_set` +/// propagated write errors, **every API returned 500**. +/// +/// Building fresh lets deadpool recycle and re-establish broken connections on +/// checkout, so caching self-heals without a process restart. The multiplexed +/// connection is cheaply cloneable (`Arc`-backed), so a per-call pool checkout +/// is negligible overhead next to the network I/O. +pub(crate) async fn fresh_redis_cache() -> Result { + let pool = redis_pool().map_err(CacheError::Io)?; + let conn = pool + .get() + .await + .map_err(|e| CacheError::Io(e.to_string()))?; + let mux = deadpool_redis::Connection::take(conn); + Ok(RedisCache::new(mux)) } /// Convenience wrappers so callers can use the mytheclipse `Cache` methods /// directly without importing the trait twice. pub async fn get(key: &str) -> Result>, CacheError> { - redis_cache().await?.get(key).await + fresh_redis_cache().await?.get(key).await } pub async fn set( @@ -46,9 +45,9 @@ pub async fn set( value: Vec, ttl: Option, ) -> Result<(), CacheError> { - redis_cache().await?.set(key, value, ttl).await + fresh_redis_cache().await?.set(key, value, ttl).await } pub async fn invalidate(key: &str) -> Result<(), CacheError> { - redis_cache().await?.invalidate(key).await + fresh_redis_cache().await?.invalidate(key).await } diff --git a/src/infrastructure/cache/redis.rs b/src/infrastructure/cache/redis.rs index cbe9ede..44afdc5 100644 --- a/src/infrastructure/cache/redis.rs +++ b/src/infrastructure/cache/redis.rs @@ -24,13 +24,13 @@ impl<'a> Cache<'a> { } } - async fn cache(&self) -> Result<&'static mytheclipse_cache::RedisCache, CacheError> { - super::mytheclipse::redis_cache().await + async fn cache(&self) -> Result { + super::mytheclipse::fresh_redis_cache().await } pub async fn get(&self, key: &str) -> Option { match self.cache().await { - Ok(cache) => match CacheTrait::get(cache, key).await { + Ok(cache) => match CacheTrait::get(&cache, key).await { Ok(Some(bytes)) => serde_json::from_slice(&bytes).ok(), Ok(None) => None, Err(e) => { @@ -69,7 +69,7 @@ impl<'a> Cache<'a> { let json = serde_json::to_vec(value).map_err(|e| e.to_string())?; let ttl = std::time::Duration::from_secs(ttl_secs); let cache = self.cache().await.map_err(|e| e.to_string())?; - CacheTrait::set(cache, key, json, Some(ttl)) + CacheTrait::set(&cache, key, json, Some(ttl)) .await .map_err(|e| e.to_string())?; debug!("Cache: set key {} with TTL {}s", key, ttl_secs); @@ -78,7 +78,7 @@ impl<'a> Cache<'a> { pub async fn delete(&self, key: &str) -> Result<(), String> { let cache = self.cache().await.map_err(|e| e.to_string())?; - CacheTrait::invalidate(cache, key) + CacheTrait::invalidate(&cache, key) .await .map_err(|e| e.to_string()) } @@ -88,6 +88,12 @@ impl<'a> Cache<'a> { } /// Get or set: returns cached value or computes and caches new value. + /// + /// The cache is best-effort: if the post-compute write fails (e.g. a + /// transient Redis outage or a broken pooled connection), the freshly + /// computed value is still returned rather than propagating a 500. Only a + /// cache *read* failure is silently tolerated; a compute failure still + /// propagates. pub async fn get_or_set( &self, key: &str, @@ -106,7 +112,12 @@ impl<'a> Cache<'a> { debug!("Cache miss: {}", key); let value = compute().await?; - self.set_with_ttl(key, &value, ttl_secs).await?; + // Best-effort write: a failure here must not fail the request — the + // value is already valid. Log and continue (read path swallows errors + // too, so a broken cache degrades to cache-less, never to 500). + if let Err(e) = self.set_with_ttl(key, &value, ttl_secs).await { + debug!("Cache: failed to write {} (non-fatal): {}", key, e); + } Ok(value) } }