aster_forge_cache/
redis_cache.rs

1//! Redis cache backend with local fallback support.
2//!
3//! Operations are bounded by short timeouts so a slow Redis instance does not stall request paths.
4//! When Redis becomes unavailable, writes are mirrored into an in-memory fallback and availability
5//! checks are rate-limited until the cooldown expires. Only connectivity failures and transient
6//! server states open the fallback circuit; deterministic command errors (for example WRONGTYPE)
7//! are logged and fall back for that single operation without degrading the backend.
8
9use super::{CacheBackend, CacheEndpoint, CacheError, Result, memory::MemoryCache};
10use async_trait::async_trait;
11use redis::{AsyncCommands, ExistenceCheck, SetExpiry, SetOptions};
12use std::future::Future;
13use std::sync::Mutex;
14use std::time::{Duration, Instant};
15
16const REDIS_CACHE_OPERATION_TIMEOUT: Duration = Duration::from_millis(250);
17const REDIS_CACHE_CONNECTION_TIMEOUT: Duration = Duration::from_millis(500);
18const REDIS_CACHE_RECONNECT_MIN_DELAY: Duration = Duration::from_millis(100);
19const REDIS_CACHE_RECONNECT_MAX_DELAY: Duration = Duration::from_millis(500);
20const REDIS_CACHE_RECONNECT_RETRIES: usize = 1;
21const REDIS_CACHE_FALLBACK_COOLDOWN: Duration = Duration::from_secs(5);
22
23/// Escapes a literal key prefix for use inside a Redis `SCAN MATCH` glob pattern.
24///
25/// Redis glob patterns treat `*`, `?`, `[...]` (including ranges and negation) as
26/// metacharacters and `\` as the escape character, so a raw prefix containing any of
27/// them would match unintended keys — and `invalidate_prefix` would then delete those
28/// keys while the intended ones survive. Escaping `\` first, then every metacharacter,
29/// makes the prefix match verbatim before the trailing `*` is appended.
30fn escape_scan_glob_literal(prefix: &str) -> String {
31    prefix
32        .replace('\\', "\\\\")
33        .replace('*', "\\*")
34        .replace('?', "\\?")
35        .replace('[', "\\[")
36        .replace(']', "\\]")
37}
38
39/// Redis cache backend with a short-lived local memory fallback.
40pub struct RedisCache {
41    inner: RedisCacheInner<RedisConnectionManager>,
42}
43
44struct RedisCacheInner<R> {
45    redis: R,
46    default_ttl: u64,
47    local: MemoryCache,
48    availability: RedisAvailability,
49}
50
51#[async_trait]
52trait RedisClient: Send + Sync {
53    async fn get(&self, key: &str) -> redis::RedisResult<Option<Vec<u8>>>;
54    async fn take(&self, key: &str) -> redis::RedisResult<Option<Vec<u8>>>;
55    async fn set_ex(&self, key: &str, value: Vec<u8>, ttl: u64) -> redis::RedisResult<()>;
56    async fn set_nx_ex(
57        &self,
58        key: &str,
59        value: Vec<u8>,
60        ttl: u64,
61    ) -> redis::RedisResult<Option<String>>;
62    async fn delete(&self, key: &str) -> redis::RedisResult<()>;
63    async fn scan_prefix(
64        &self,
65        cursor: u64,
66        pattern: &str,
67    ) -> redis::RedisResult<(u64, Vec<String>)>;
68    async fn delete_keys(&self, keys: &[String]) -> redis::RedisResult<()>;
69    async fn ping(&self) -> redis::RedisResult<String>;
70}
71
72struct RedisConnectionManager {
73    conn: redis::aio::ConnectionManager,
74}
75
76impl RedisConnectionManager {
77    fn new(conn: redis::aio::ConnectionManager) -> Self {
78        Self { conn }
79    }
80}
81
82#[async_trait]
83impl RedisClient for RedisConnectionManager {
84    async fn get(&self, key: &str) -> redis::RedisResult<Option<Vec<u8>>> {
85        let mut conn = self.conn.clone();
86        conn.get::<_, Option<Vec<u8>>>(key).await
87    }
88
89    async fn take(&self, key: &str) -> redis::RedisResult<Option<Vec<u8>>> {
90        let mut conn = self.conn.clone();
91        redis::Script::new(
92            r#"
93            local value = redis.call("GET", KEYS[1])
94            if value then
95                redis.call("DEL", KEYS[1])
96            end
97            return value
98            "#,
99        )
100        .key(key)
101        .invoke_async::<Option<Vec<u8>>>(&mut conn)
102        .await
103    }
104
105    async fn set_ex(&self, key: &str, value: Vec<u8>, ttl: u64) -> redis::RedisResult<()> {
106        let mut conn = self.conn.clone();
107        conn.set_ex::<_, _, ()>(key, value, ttl).await
108    }
109
110    async fn set_nx_ex(
111        &self,
112        key: &str,
113        value: Vec<u8>,
114        ttl: u64,
115    ) -> redis::RedisResult<Option<String>> {
116        let options = SetOptions::default()
117            .conditional_set(ExistenceCheck::NX)
118            .with_expiration(SetExpiry::EX(ttl));
119        let mut conn = self.conn.clone();
120        conn.set_options::<_, _, Option<String>>(key, value, options)
121            .await
122    }
123
124    async fn delete(&self, key: &str) -> redis::RedisResult<()> {
125        let mut conn = self.conn.clone();
126        conn.del::<_, ()>(key).await
127    }
128
129    async fn scan_prefix(
130        &self,
131        cursor: u64,
132        pattern: &str,
133    ) -> redis::RedisResult<(u64, Vec<String>)> {
134        let mut conn = self.conn.clone();
135        let mut scan_cmd = redis::cmd("SCAN");
136        scan_cmd
137            .arg(cursor)
138            .arg("MATCH")
139            .arg(pattern)
140            .arg("COUNT")
141            .arg(100)
142            .query_async::<(u64, Vec<String>)>(&mut conn)
143            .await
144    }
145
146    async fn delete_keys(&self, keys: &[String]) -> redis::RedisResult<()> {
147        let mut conn = self.conn.clone();
148        conn.del::<_, ()>(keys).await
149    }
150
151    async fn ping(&self) -> redis::RedisResult<String> {
152        let mut conn = self.conn.clone();
153        redis::cmd("PING").query_async::<String>(&mut conn).await
154    }
155}
156
157impl RedisCache {
158    /// Creates a Redis cache from a Redis URL and default TTL in seconds.
159    ///
160    /// # Errors
161    ///
162    /// Returns [`CacheError`] when the URL is invalid or the initial Redis connection fails.
163    pub async fn new(url: &str, default_ttl: u64) -> Result<Self> {
164        let client = redis::Client::open(url)?;
165        Self::from_client(client, default_ttl).await
166    }
167
168    /// Creates a Redis cache from a base URL and raw credentials.
169    ///
170    /// # Errors
171    ///
172    /// Returns [`CacheError`] when credential injection or URL parsing fails, or Redis cannot be
173    /// reached during initialization.
174    pub async fn from_credentials(
175        base_url: &str,
176        username: Option<&str>,
177        password: Option<&str>,
178        default_ttl: u64,
179    ) -> Result<Self> {
180        let url = aster_forge_utils::url::url_with_credentials(
181            base_url,
182            username,
183            password,
184            "Redis cache base URL",
185        )
186        .map_err(|error| CacheError::InvalidConfiguration(error.to_string()))?;
187        let client = redis::Client::open(url).map_err(|_| {
188            CacheError::InvalidConfiguration(
189                "invalid Redis cache connection configuration".to_string(),
190            )
191        })?;
192        Self::from_client(client, default_ttl).await
193    }
194
195    pub(crate) async fn from_endpoint(endpoint: &CacheEndpoint, default_ttl: u64) -> Result<Self> {
196        match endpoint {
197            CacheEndpoint::Url(url) => Self::new(url, default_ttl).await,
198            CacheEndpoint::Credentials {
199                base_url,
200                username,
201                password,
202            } => {
203                Self::from_credentials(
204                    base_url,
205                    username.as_deref(),
206                    password.as_deref(),
207                    default_ttl,
208                )
209                .await
210            }
211        }
212    }
213
214    async fn from_client(client: redis::Client, default_ttl: u64) -> Result<Self> {
215        let manager_config = redis::aio::ConnectionManagerConfig::new()
216            .set_response_timeout(Some(REDIS_CACHE_OPERATION_TIMEOUT))
217            .set_connection_timeout(Some(REDIS_CACHE_CONNECTION_TIMEOUT))
218            .set_min_delay(REDIS_CACHE_RECONNECT_MIN_DELAY)
219            .set_max_delay(REDIS_CACHE_RECONNECT_MAX_DELAY)
220            .set_number_of_retries(REDIS_CACHE_RECONNECT_RETRIES);
221        let conn = redis::aio::ConnectionManager::new_with_config(client, manager_config).await?;
222        Ok(Self {
223            inner: RedisCacheInner::new(RedisConnectionManager::new(conn), default_ttl),
224        })
225    }
226}
227
228impl<R> RedisCacheInner<R>
229where
230    R: RedisClient,
231{
232    fn new(redis: R, default_ttl: u64) -> Self {
233        Self {
234            redis,
235            default_ttl,
236            local: MemoryCache::new(default_ttl),
237            availability: RedisAvailability::default(),
238        }
239    }
240
241    async fn get_local_bytes(&self, key: &str) -> Option<Vec<u8>> {
242        self.local.get_bytes(key).await
243    }
244
245    async fn set_local_bytes(&self, key: &str, value: Vec<u8>, ttl_secs: Option<u64>) {
246        self.local.set_bytes(key, value, ttl_secs).await;
247    }
248
249    async fn set_local_bytes_if_absent(
250        &self,
251        key: &str,
252        value: Vec<u8>,
253        ttl_secs: Option<u64>,
254    ) -> bool {
255        self.local.set_bytes_if_absent(key, value, ttl_secs).await
256    }
257
258    async fn delete_local(&self, key: &str) {
259        self.local.delete(key).await;
260    }
261
262    async fn invalidate_local_prefix(&self, prefix: &str) {
263        self.local.invalidate_prefix(prefix).await;
264    }
265
266    async fn redis_operation<T, Fut>(&self, operation: &'static str, future: Fut) -> Option<T>
267    where
268        T: Send,
269        Fut: Future<Output = redis::RedisResult<T>> + Send,
270    {
271        if let Some(remaining) = self.redis_unavailable_for() {
272            tracing::trace!(
273                operation,
274                remaining_ms = duration_millis_u64(remaining),
275                "redis cache circuit open; skipping redis operation"
276            );
277            return None;
278        }
279
280        match tokio::time::timeout(REDIS_CACHE_OPERATION_TIMEOUT, future).await {
281            Ok(Ok(value)) => {
282                self.mark_redis_success(operation);
283                Some(value)
284            }
285            Ok(Err(error)) => {
286                self.mark_redis_error(operation, &error);
287                None
288            }
289            Err(_) => {
290                self.mark_redis_timeout(operation);
291                None
292            }
293        }
294    }
295
296    fn redis_unavailable_for(&self) -> Option<Duration> {
297        self.availability.unavailable_for(Instant::now())
298    }
299
300    fn mark_redis_success(&self, operation: &'static str) {
301        if self.availability.mark_success() {
302            tracing::info!(operation, "redis cache recovered; closing fallback circuit");
303        }
304    }
305
306    fn mark_redis_error(&self, operation: &'static str, error: &redis::RedisError) {
307        if !redis_error_indicates_unavailability(error) {
308            tracing::warn!(
309                operation,
310                error = %error,
311                "redis cache command error; leaving fallback circuit closed"
312            );
313            return;
314        }
315        if self
316            .availability
317            .mark_failure(Instant::now(), REDIS_CACHE_FALLBACK_COOLDOWN)
318        {
319            tracing::warn!(
320                operation,
321                error = %error,
322                cooldown_secs = REDIS_CACHE_FALLBACK_COOLDOWN.as_secs(),
323                "redis cache unavailable; using local fallback temporarily"
324            );
325        } else {
326            tracing::debug!(
327                operation,
328                error = %error,
329                "redis cache operation failed while fallback circuit is already open"
330            );
331        }
332    }
333
334    fn mark_redis_timeout(&self, operation: &'static str) {
335        if self
336            .availability
337            .mark_failure(Instant::now(), REDIS_CACHE_FALLBACK_COOLDOWN)
338        {
339            tracing::warn!(
340                operation,
341                timeout_ms = duration_millis_u64(REDIS_CACHE_OPERATION_TIMEOUT),
342                cooldown_secs = REDIS_CACHE_FALLBACK_COOLDOWN.as_secs(),
343                "redis cache operation timed out; using local fallback temporarily"
344            );
345        } else {
346            tracing::debug!(
347                operation,
348                timeout_ms = duration_millis_u64(REDIS_CACHE_OPERATION_TIMEOUT),
349                "redis cache operation timed out while fallback circuit is already open"
350            );
351        }
352    }
353}
354
355impl<R> RedisCacheInner<R>
356where
357    R: RedisClient,
358{
359    async fn health_check(&self) -> Result<()> {
360        if let Some(remaining) = self.redis_unavailable_for() {
361            return Err(CacheError::RedisFallbackMode {
362                remaining_ms: remaining.as_millis(),
363            });
364        }
365
366        match tokio::time::timeout(REDIS_CACHE_OPERATION_TIMEOUT, self.redis.ping()).await {
367            Ok(Ok(_)) => {
368                self.mark_redis_success("health_check");
369                Ok(())
370            }
371            Ok(Err(error)) => {
372                self.mark_redis_error("health_check", &error);
373                Err(CacheError::RedisHealthCheck(error.to_string()))
374            }
375            Err(_) => {
376                self.mark_redis_timeout("health_check");
377                Err(CacheError::RedisHealthCheckTimeout {
378                    timeout_ms: REDIS_CACHE_OPERATION_TIMEOUT.as_millis(),
379                })
380            }
381        }
382    }
383
384    async fn get_bytes(&self, key: &str) -> Option<Vec<u8>> {
385        match self.redis_operation("get", self.redis.get(key)).await {
386            // A successful Redis read makes Redis authoritative for this key: a local
387            // shadow left by a previous outage is redundant at best and stale at worst,
388            // and keeping it would resurrect never-persisted data during the next outage.
389            Some(value) => {
390                self.delete_local(key).await;
391                value
392            }
393            None => self.get_local_bytes(key).await,
394        }
395    }
396
397    async fn take_bytes(&self, key: &str) -> Option<Vec<u8>> {
398        match self.redis_operation("take", self.redis.take(key)).await {
399            Some(value) => {
400                self.delete_local(key).await;
401                value
402            }
403            None => self.local.take_bytes(key).await,
404        }
405    }
406
407    async fn set_bytes(&self, key: &str, value: Vec<u8>, ttl_secs: Option<u64>) {
408        let ttl = ttl_secs.unwrap_or(self.default_ttl);
409        if ttl == 0 {
410            // Redis rejects SETEX with a zero TTL. An immediate delete produces the
411            // documented "expires immediately" observable state without issuing an
412            // invalid command.
413            self.delete(key).await;
414            return;
415        }
416        if self
417            .redis_operation("set", self.redis.set_ex(key, value.clone(), ttl))
418            .await
419            .is_some()
420        {
421            self.delete_local(key).await;
422        } else {
423            self.set_local_bytes(key, value, ttl_secs).await;
424        }
425    }
426
427    async fn set_bytes_if_absent(&self, key: &str, value: Vec<u8>, ttl_secs: Option<u64>) -> bool {
428        let ttl = ttl_secs.unwrap_or(self.default_ttl);
429        if ttl == 0 {
430            // A zero-TTL insert expires immediately, so the outcome only depends on
431            // whether a live value exists; nothing is retained either way.
432            return match self
433                .redis_operation("set_if_absent", self.redis.get(key))
434                .await
435            {
436                Some(existing) => {
437                    self.delete_local(key).await;
438                    existing.is_none()
439                }
440                None => self.set_local_bytes_if_absent(key, value, Some(0)).await,
441            };
442        }
443        match self
444            .redis_operation(
445                "set_if_absent",
446                self.redis.set_nx_ex(key, value.clone(), ttl),
447            )
448            .await
449        {
450            Some(Some(_)) => {
451                self.delete_local(key).await;
452                true
453            }
454            Some(None) => {
455                self.delete_local(key).await;
456                false
457            }
458            None => self.set_local_bytes_if_absent(key, value, ttl_secs).await,
459        }
460    }
461
462    async fn delete(&self, key: &str) {
463        self.delete_local(key).await;
464        let _: Option<()> = self.redis_operation("delete", self.redis.delete(key)).await;
465    }
466
467    async fn delete_many(&self, keys: &[String]) {
468        for key in keys {
469            self.delete_local(key).await;
470        }
471        if !keys.is_empty() {
472            let _: Option<()> = self
473                .redis_operation("delete_many", self.redis.delete_keys(keys))
474                .await;
475        }
476    }
477
478    async fn invalidate_prefix(&self, prefix: &str) {
479        self.invalidate_local_prefix(prefix).await;
480
481        let pattern = format!("{}*", escape_scan_glob_literal(prefix));
482        let mut cursor: u64 = 0;
483        loop {
484            let Some((next_cursor, keys)) = self
485                .redis_operation(
486                    "invalidate_prefix_scan",
487                    self.redis.scan_prefix(cursor, &pattern),
488                )
489                .await
490            else {
491                break;
492            };
493            if !keys.is_empty()
494                && self
495                    .redis_operation("invalidate_prefix_delete", self.redis.delete_keys(&keys))
496                    .await
497                    .is_none()
498            {
499                break;
500            }
501            cursor = next_cursor;
502            if cursor == 0 {
503                break;
504            }
505        }
506    }
507}
508
509#[async_trait]
510impl CacheBackend for RedisCache {
511    fn backend_name(&self) -> &'static str {
512        "redis"
513    }
514
515    async fn health_check(&self) -> Result<()> {
516        self.inner.health_check().await
517    }
518
519    async fn get_bytes(&self, key: &str) -> Option<Vec<u8>> {
520        self.inner.get_bytes(key).await
521    }
522
523    async fn take_bytes(&self, key: &str) -> Option<Vec<u8>> {
524        self.inner.take_bytes(key).await
525    }
526
527    async fn set_bytes(&self, key: &str, value: Vec<u8>, ttl_secs: Option<u64>) {
528        self.inner.set_bytes(key, value, ttl_secs).await;
529    }
530
531    async fn set_bytes_if_absent(&self, key: &str, value: Vec<u8>, ttl_secs: Option<u64>) -> bool {
532        self.inner.set_bytes_if_absent(key, value, ttl_secs).await
533    }
534
535    async fn delete(&self, key: &str) {
536        self.inner.delete(key).await;
537    }
538
539    async fn delete_many(&self, keys: &[String]) {
540        self.inner.delete_many(keys).await;
541    }
542
543    async fn invalidate_prefix(&self, prefix: &str) {
544        self.inner.invalidate_prefix(prefix).await;
545    }
546}
547
548fn duration_millis_u64(duration: Duration) -> u64 {
549    u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
550}
551
552/// Returns whether a Redis error indicates the server is unreachable or in a transient
553/// state, as opposed to a deterministic command, data, or configuration error that a
554/// fallback circuit cannot fix.
555fn redis_error_indicates_unavailability(error: &redis::RedisError) -> bool {
556    match error.kind() {
557        redis::ErrorKind::Io | redis::ErrorKind::ClusterConnectionNotFound => true,
558        redis::ErrorKind::Server(kind) => matches!(
559            kind,
560            redis::ServerErrorKind::BusyLoading
561                | redis::ServerErrorKind::TryAgain
562                | redis::ServerErrorKind::ClusterDown
563                | redis::ServerErrorKind::MasterDown
564                | redis::ServerErrorKind::ReadOnly
565        ),
566        _ => false,
567    }
568}
569
570#[derive(Default)]
571struct RedisAvailability {
572    unavailable_until: Mutex<Option<Instant>>,
573}
574
575impl RedisAvailability {
576    fn unavailable_for(&self, now: Instant) -> Option<Duration> {
577        let mut unavailable_until = self.lock_unavailable_until();
578        match *unavailable_until {
579            Some(deadline) if deadline > now => Some(deadline.duration_since(now)),
580            Some(_) => {
581                *unavailable_until = None;
582                None
583            }
584            None => None,
585        }
586    }
587
588    fn mark_failure(&self, now: Instant, cooldown: Duration) -> bool {
589        let mut unavailable_until = self.lock_unavailable_until();
590        let was_available = unavailable_until.is_none_or(|deadline| deadline <= now);
591        *unavailable_until = now.checked_add(cooldown).or(Some(now));
592        was_available
593    }
594
595    fn mark_success(&self) -> bool {
596        self.lock_unavailable_until().take().is_some()
597    }
598
599    fn lock_unavailable_until(&self) -> std::sync::MutexGuard<'_, Option<Instant>> {
600        self.unavailable_until
601            .lock()
602            .unwrap_or_else(std::sync::PoisonError::into_inner)
603    }
604}
605
606#[cfg(test)]
607mod tests {
608    use super::{
609        CacheBackend, REDIS_CACHE_FALLBACK_COOLDOWN, RedisAvailability, RedisCacheInner,
610        RedisClient, escape_scan_glob_literal,
611    };
612    use async_trait::async_trait;
613    use std::collections::HashMap;
614    use std::sync::{
615        Arc, Mutex,
616        atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering},
617    };
618    use std::time::{Duration, Instant};
619    use tokio::time::sleep;
620
621    #[derive(Default)]
622    struct FakeRedisClient {
623        entries: Mutex<HashMap<String, Vec<u8>>>,
624        scan_pages: Mutex<HashMap<u64, Vec<String>>>,
625        fail_operations: AtomicBool,
626        fail_command_errors: AtomicBool,
627        next_scan_cursor: AtomicU64,
628        get_calls: AtomicUsize,
629        take_calls: AtomicUsize,
630        set_calls: AtomicUsize,
631        set_nx_calls: AtomicUsize,
632        delete_calls: AtomicUsize,
633        scan_calls: AtomicUsize,
634        delete_keys_calls: AtomicUsize,
635        ping_calls: AtomicUsize,
636    }
637
638    impl FakeRedisClient {
639        fn set_fail_operations(&self, fail: bool) {
640            self.fail_operations.store(fail, Ordering::SeqCst);
641        }
642
643        fn set_fail_command_errors(&self, fail: bool) {
644            self.fail_command_errors.store(fail, Ordering::SeqCst);
645        }
646
647        fn insert(&self, key: &str, value: &[u8]) {
648            self.entries
649                .lock()
650                .unwrap_or_else(std::sync::PoisonError::into_inner)
651                .insert(key.to_string(), value.to_vec());
652        }
653
654        fn contains_key(&self, key: &str) -> bool {
655            self.entries
656                .lock()
657                .unwrap_or_else(std::sync::PoisonError::into_inner)
658                .contains_key(key)
659        }
660
661        fn get_call_count(&self) -> usize {
662            self.get_calls.load(Ordering::SeqCst)
663        }
664
665        fn take_call_count(&self) -> usize {
666            self.take_calls.load(Ordering::SeqCst)
667        }
668
669        fn set_call_count(&self) -> usize {
670            self.set_calls.load(Ordering::SeqCst)
671        }
672
673        fn set_nx_call_count(&self) -> usize {
674            self.set_nx_calls.load(Ordering::SeqCst)
675        }
676
677        fn delete_call_count(&self) -> usize {
678            self.delete_calls.load(Ordering::SeqCst)
679        }
680
681        fn scan_call_count(&self) -> usize {
682            self.scan_calls.load(Ordering::SeqCst)
683        }
684
685        fn delete_keys_call_count(&self) -> usize {
686            self.delete_keys_calls.load(Ordering::SeqCst)
687        }
688
689        fn ping_call_count(&self) -> usize {
690            self.ping_calls.load(Ordering::SeqCst)
691        }
692
693        fn maybe_fail(&self) -> redis::RedisResult<()> {
694            if self.fail_operations.load(Ordering::SeqCst) {
695                Err(redis::RedisError::from((
696                    redis::ErrorKind::Io,
697                    "fake redis unavailable",
698                )))
699            } else if self.fail_command_errors.load(Ordering::SeqCst) {
700                Err(redis::RedisError::from((
701                    redis::ErrorKind::Server(redis::ServerErrorKind::ResponseError),
702                    "ERR invalid expire time in 'setex' command",
703                )))
704            } else {
705                Ok(())
706            }
707        }
708    }
709
710    #[async_trait]
711    impl RedisClient for Arc<FakeRedisClient> {
712        async fn get(&self, key: &str) -> redis::RedisResult<Option<Vec<u8>>> {
713            self.get_calls.fetch_add(1, Ordering::SeqCst);
714            self.maybe_fail()?;
715            Ok(self
716                .entries
717                .lock()
718                .unwrap_or_else(std::sync::PoisonError::into_inner)
719                .get(key)
720                .cloned())
721        }
722
723        async fn take(&self, key: &str) -> redis::RedisResult<Option<Vec<u8>>> {
724            self.take_calls.fetch_add(1, Ordering::SeqCst);
725            self.maybe_fail()?;
726            Ok(self
727                .entries
728                .lock()
729                .unwrap_or_else(std::sync::PoisonError::into_inner)
730                .remove(key))
731        }
732
733        async fn set_ex(&self, key: &str, value: Vec<u8>, _ttl: u64) -> redis::RedisResult<()> {
734            self.set_calls.fetch_add(1, Ordering::SeqCst);
735            self.maybe_fail()?;
736            self.entries
737                .lock()
738                .unwrap_or_else(std::sync::PoisonError::into_inner)
739                .insert(key.to_string(), value);
740            Ok(())
741        }
742
743        async fn set_nx_ex(
744            &self,
745            key: &str,
746            value: Vec<u8>,
747            _ttl: u64,
748        ) -> redis::RedisResult<Option<String>> {
749            self.set_nx_calls.fetch_add(1, Ordering::SeqCst);
750            self.maybe_fail()?;
751            let mut entries = self
752                .entries
753                .lock()
754                .unwrap_or_else(std::sync::PoisonError::into_inner);
755            if entries.contains_key(key) {
756                Ok(None)
757            } else {
758                entries.insert(key.to_string(), value);
759                Ok(Some("OK".to_string()))
760            }
761        }
762
763        async fn delete(&self, key: &str) -> redis::RedisResult<()> {
764            self.delete_calls.fetch_add(1, Ordering::SeqCst);
765            self.maybe_fail()?;
766            self.entries
767                .lock()
768                .unwrap_or_else(std::sync::PoisonError::into_inner)
769                .remove(key);
770            Ok(())
771        }
772
773        async fn scan_prefix(
774            &self,
775            cursor: u64,
776            pattern: &str,
777        ) -> redis::RedisResult<(u64, Vec<String>)> {
778            const PAGE_SIZE: usize = 2;
779
780            self.scan_calls.fetch_add(1, Ordering::SeqCst);
781            self.maybe_fail()?;
782            let prefix = pattern
783                .strip_suffix('*')
784                .expect("prefix invalidation should scan a trailing-star pattern");
785            let mut keys = if cursor == 0 {
786                let mut keys: Vec<String> = self
787                    .entries
788                    .lock()
789                    .unwrap_or_else(std::sync::PoisonError::into_inner)
790                    .keys()
791                    .filter(|key| key.starts_with(prefix))
792                    .cloned()
793                    .collect();
794                keys.sort();
795                keys
796            } else {
797                self.scan_pages
798                    .lock()
799                    .unwrap_or_else(std::sync::PoisonError::into_inner)
800                    .remove(&cursor)
801                    .unwrap_or_default()
802            };
803            if keys.is_empty() {
804                return Ok((0, Vec::new()));
805            }
806            let page_len = PAGE_SIZE.min(keys.len());
807            let remaining = keys.split_off(page_len);
808            let page = keys;
809            let next_cursor = if remaining.is_empty() {
810                0
811            } else {
812                let next_cursor = self.next_scan_cursor.fetch_add(1, Ordering::SeqCst) + 1;
813                self.scan_pages
814                    .lock()
815                    .unwrap_or_else(std::sync::PoisonError::into_inner)
816                    .insert(next_cursor, remaining);
817                next_cursor
818            };
819            Ok((next_cursor, page))
820        }
821
822        async fn delete_keys(&self, keys: &[String]) -> redis::RedisResult<()> {
823            self.delete_keys_calls.fetch_add(1, Ordering::SeqCst);
824            self.maybe_fail()?;
825            let mut entries = self
826                .entries
827                .lock()
828                .unwrap_or_else(std::sync::PoisonError::into_inner);
829            for key in keys {
830                entries.remove(key);
831            }
832            Ok(())
833        }
834
835        async fn ping(&self) -> redis::RedisResult<String> {
836            self.ping_calls.fetch_add(1, Ordering::SeqCst);
837            self.maybe_fail()?;
838            Ok("PONG".to_string())
839        }
840    }
841
842    fn cache_with_fake_redis(
843        default_ttl: u64,
844    ) -> (RedisCacheInner<Arc<FakeRedisClient>>, Arc<FakeRedisClient>) {
845        let redis = Arc::new(FakeRedisClient::default());
846        (RedisCacheInner::new(redis.clone(), default_ttl), redis)
847    }
848
849    fn open_fallback_circuit<R: RedisClient>(cache: &RedisCacheInner<R>) {
850        assert!(
851            cache
852                .availability
853                .mark_failure(Instant::now(), REDIS_CACHE_FALLBACK_COOLDOWN)
854        );
855    }
856
857    #[test]
858    fn redis_availability_skips_until_cooldown_expires() {
859        let availability = RedisAvailability::default();
860        let now = Instant::now();
861
862        assert!(availability.unavailable_for(now).is_none());
863        assert!(availability.mark_failure(now, Duration::from_secs(5)));
864        assert_eq!(
865            availability.unavailable_for(now + Duration::from_secs(2)),
866            Some(Duration::from_secs(3))
867        );
868        assert!(
869            availability
870                .unavailable_for(now + Duration::from_secs(6))
871                .is_none()
872        );
873    }
874
875    #[test]
876    fn redis_availability_reports_recovery_once() {
877        let availability = RedisAvailability::default();
878        let now = Instant::now();
879
880        assert!(availability.mark_failure(now, Duration::from_secs(5)));
881        assert!(availability.mark_success());
882        assert!(!availability.mark_success());
883    }
884
885    #[test]
886    fn redis_availability_repeated_failures_only_report_transition_once() {
887        let availability = RedisAvailability::default();
888        let now = Instant::now();
889
890        assert!(availability.mark_failure(now, Duration::from_secs(5)));
891        assert!(!availability.mark_failure(now + Duration::from_secs(1), Duration::from_secs(5)));
892    }
893
894    #[tokio::test]
895    async fn fallback_set_and_get_round_trip_while_circuit_is_open() {
896        let (cache, redis) = cache_with_fake_redis(60);
897        open_fallback_circuit(&cache);
898
899        cache.set_bytes("ticket", b"local".to_vec(), Some(60)).await;
900
901        assert_eq!(cache.get_bytes("ticket").await, Some(b"local".to_vec()));
902        assert_eq!(
903            redis.set_call_count(),
904            0,
905            "circuit-open set should skip Redis"
906        );
907        assert_eq!(
908            redis.get_call_count(),
909            0,
910            "circuit-open get should skip Redis"
911        );
912    }
913
914    #[tokio::test]
915    async fn failed_redis_set_stores_value_in_local_fallback() {
916        let (cache, redis) = cache_with_fake_redis(60);
917        redis.set_fail_operations(true);
918
919        cache
920            .set_bytes("session", b"fallback".to_vec(), Some(60))
921            .await;
922
923        assert_eq!(redis.set_call_count(), 1);
924        assert_eq!(cache.get_bytes("session").await, Some(b"fallback".to_vec()));
925        assert_eq!(
926            redis.get_call_count(),
927            0,
928            "first failed set opens the circuit, so later get should skip Redis"
929        );
930    }
931
932    #[tokio::test]
933    async fn redis_miss_does_not_return_stale_local_value_when_redis_is_available() {
934        let (cache, redis) = cache_with_fake_redis(60);
935        open_fallback_circuit(&cache);
936        cache
937            .set_bytes("snapshot", b"stale-local".to_vec(), Some(60))
938            .await;
939        redis.set_fail_operations(false);
940        cache.availability.mark_success();
941
942        assert_eq!(cache.get_bytes("snapshot").await, None);
943        assert_eq!(redis.get_call_count(), 1);
944    }
945
946    #[tokio::test]
947    async fn successful_redis_get_clears_local_shadow_so_it_cannot_resurrect() {
948        let (cache, redis) = cache_with_fake_redis(60);
949        // Outage: the write lands only in the local fallback.
950        open_fallback_circuit(&cache);
951        cache
952            .set_bytes("snapshot", b"local-only".to_vec(), Some(60))
953            .await;
954        redis.set_fail_operations(false);
955        cache.availability.mark_success();
956
957        // Redis answers with a miss: the shadow must be deleted, not just bypassed.
958        assert_eq!(cache.get_bytes("snapshot").await, None);
959        assert_eq!(cache.local.get_bytes("snapshot").await, None);
960
961        // The next outage must not resurrect the never-persisted value.
962        open_fallback_circuit(&cache);
963        assert_eq!(cache.get_bytes("snapshot").await, None);
964    }
965
966    #[tokio::test]
967    async fn successful_redis_hit_clears_divergent_local_shadow() {
968        let (cache, redis) = cache_with_fake_redis(60);
969        redis.insert("profile", b"redis-value");
970        open_fallback_circuit(&cache);
971        cache
972            .set_bytes("profile", b"local-shadow".to_vec(), Some(60))
973            .await;
974        redis.set_fail_operations(false);
975        cache.availability.mark_success();
976
977        assert_eq!(
978            cache.get_bytes("profile").await,
979            Some(b"redis-value".to_vec())
980        );
981        assert_eq!(cache.local.get_bytes("profile").await, None);
982    }
983
984    #[tokio::test]
985    async fn successful_redis_set_clears_local_fallback_shadow() {
986        let (cache, redis) = cache_with_fake_redis(60);
987        open_fallback_circuit(&cache);
988        cache
989            .set_bytes("profile", b"local-shadow".to_vec(), Some(60))
990            .await;
991        cache.availability.mark_success();
992
993        cache
994            .set_bytes("profile", b"redis-value".to_vec(), Some(60))
995            .await;
996
997        assert_eq!(redis.set_call_count(), 1);
998        assert!(redis.contains_key("profile"));
999        assert_eq!(cache.local.get_bytes("profile").await, None);
1000        assert_eq!(
1001            cache.get_bytes("profile").await,
1002            Some(b"redis-value".to_vec())
1003        );
1004    }
1005
1006    #[tokio::test]
1007    async fn take_bytes_consumes_redis_entry_atomically() {
1008        let (cache, redis) = cache_with_fake_redis(60);
1009        redis.insert("challenge", b"value");
1010
1011        assert_eq!(cache.take_bytes("challenge").await, Some(b"value".to_vec()));
1012        assert_eq!(cache.take_bytes("challenge").await, None);
1013        assert!(!redis.contains_key("challenge"));
1014        assert_eq!(redis.take_call_count(), 2);
1015        assert_eq!(
1016            redis.get_call_count(),
1017            0,
1018            "take should not read through a separate GET"
1019        );
1020    }
1021
1022    #[tokio::test]
1023    async fn take_bytes_consumes_local_fallback_when_circuit_is_open() {
1024        let (cache, redis) = cache_with_fake_redis(60);
1025        open_fallback_circuit(&cache);
1026        cache
1027            .set_bytes("challenge", b"local".to_vec(), Some(60))
1028            .await;
1029
1030        assert_eq!(cache.take_bytes("challenge").await, Some(b"local".to_vec()));
1031        assert_eq!(cache.take_bytes("challenge").await, None);
1032        assert_eq!(
1033            redis.take_call_count(),
1034            0,
1035            "circuit-open take should skip Redis"
1036        );
1037    }
1038
1039    #[tokio::test]
1040    async fn take_bytes_falls_back_to_local_without_dropping_value_on_redis_failure() {
1041        let (cache, redis) = cache_with_fake_redis(60);
1042        open_fallback_circuit(&cache);
1043        cache
1044            .set_bytes("challenge", b"local".to_vec(), Some(60))
1045            .await;
1046        cache.availability.mark_success();
1047        redis.set_fail_operations(true);
1048
1049        assert_eq!(cache.take_bytes("challenge").await, Some(b"local".to_vec()));
1050        assert_eq!(cache.take_bytes("challenge").await, None);
1051        assert_eq!(redis.take_call_count(), 1);
1052    }
1053
1054    #[tokio::test]
1055    async fn fallback_set_if_absent_stores_value_and_rejects_second_insert() {
1056        let (cache, redis) = cache_with_fake_redis(60);
1057        open_fallback_circuit(&cache);
1058
1059        assert!(
1060            cache
1061                .set_bytes_if_absent("nonce", b"first".to_vec(), Some(60))
1062                .await
1063        );
1064        assert!(
1065            !cache
1066                .set_bytes_if_absent("nonce", b"second".to_vec(), Some(60))
1067                .await
1068        );
1069
1070        assert_eq!(cache.get_bytes("nonce").await, Some(b"first".to_vec()));
1071        assert_eq!(redis.set_nx_call_count(), 0);
1072    }
1073
1074    #[tokio::test]
1075    async fn fallback_set_if_absent_is_atomic_for_concurrent_callers() {
1076        let (cache, redis) = cache_with_fake_redis(60);
1077        open_fallback_circuit(&cache);
1078        let cache = Arc::new(cache);
1079        let mut tasks = Vec::new();
1080
1081        for index in 0..32 {
1082            let cache = cache.clone();
1083            tasks.push(tokio::spawn(async move {
1084                cache
1085                    .set_bytes_if_absent(
1086                        "concurrent-nonce",
1087                        format!("value-{index}").into_bytes(),
1088                        Some(60),
1089                    )
1090                    .await
1091            }));
1092        }
1093
1094        let inserted = futures::future::join_all(tasks)
1095            .await
1096            .into_iter()
1097            .map(|result| result.expect("fallback reservation task should not panic"))
1098            .filter(|inserted| *inserted)
1099            .count();
1100
1101        assert_eq!(inserted, 1);
1102        assert!(cache.get_bytes("concurrent-nonce").await.is_some());
1103        assert_eq!(redis.set_nx_call_count(), 0);
1104    }
1105
1106    #[tokio::test]
1107    async fn fallback_entries_respect_zero_ttl_boundary() {
1108        let (cache, _redis) = cache_with_fake_redis(60);
1109        open_fallback_circuit(&cache);
1110
1111        cache.set_bytes("expired", b"value".to_vec(), Some(0)).await;
1112        assert_eq!(cache.get_bytes("expired").await, None);
1113
1114        assert!(
1115            cache
1116                .set_bytes_if_absent("zero-nonce", b"first".to_vec(), Some(0))
1117                .await
1118        );
1119        assert_eq!(cache.get_bytes("zero-nonce").await, None);
1120        assert!(
1121            cache
1122                .set_bytes_if_absent("zero-nonce", b"second".to_vec(), Some(0))
1123                .await
1124        );
1125    }
1126
1127    #[tokio::test]
1128    async fn fallback_entries_expire_after_configured_ttl() {
1129        let (cache, _redis) = cache_with_fake_redis(60);
1130        open_fallback_circuit(&cache);
1131
1132        cache
1133            .set_bytes("short-lived", b"value".to_vec(), Some(1))
1134            .await;
1135        assert_eq!(
1136            cache.get_bytes("short-lived").await,
1137            Some(b"value".to_vec())
1138        );
1139
1140        sleep(Duration::from_millis(1_100)).await;
1141
1142        assert_eq!(cache.get_bytes("short-lived").await, None);
1143    }
1144
1145    #[tokio::test]
1146    async fn delete_clears_local_fallback_even_when_redis_is_unavailable() {
1147        let (cache, redis) = cache_with_fake_redis(60);
1148        open_fallback_circuit(&cache);
1149        cache
1150            .set_bytes("delete-me", b"value".to_vec(), Some(60))
1151            .await;
1152
1153        cache.delete("delete-me").await;
1154
1155        assert_eq!(cache.get_bytes("delete-me").await, None);
1156        assert_eq!(
1157            redis.delete_call_count(),
1158            0,
1159            "circuit-open delete should skip Redis"
1160        );
1161    }
1162
1163    #[tokio::test]
1164    async fn delete_many_removes_requested_redis_and_local_entries_in_one_batch() {
1165        let (cache, redis) = cache_with_fake_redis(60);
1166        redis.insert("remove:1", b"one");
1167        redis.insert("remove:2", b"two");
1168        redis.insert("keep", b"keep");
1169        open_fallback_circuit(&cache);
1170        cache
1171            .set_bytes("remove:local", b"local".to_vec(), Some(60))
1172            .await;
1173        cache.availability.mark_success();
1174
1175        cache
1176            .delete_many(&[
1177                "remove:1".to_string(),
1178                "remove:2".to_string(),
1179                "remove:local".to_string(),
1180                "missing".to_string(),
1181            ])
1182            .await;
1183        cache.delete_many(&[]).await;
1184
1185        assert!(!redis.contains_key("remove:1"));
1186        assert!(!redis.contains_key("remove:2"));
1187        assert!(redis.contains_key("keep"));
1188        assert_eq!(cache.local.get_bytes("remove:local").await, None);
1189        assert_eq!(redis.delete_keys_call_count(), 1);
1190    }
1191
1192    #[tokio::test]
1193    async fn delete_many_clears_local_only_when_circuit_is_open() {
1194        let (cache, redis) = cache_with_fake_redis(60);
1195        open_fallback_circuit(&cache);
1196        cache
1197            .set_bytes("remove:local", b"local".to_vec(), Some(60))
1198            .await;
1199
1200        cache.delete_many(&["remove:local".to_string()]).await;
1201
1202        assert_eq!(cache.get_bytes("remove:local").await, None);
1203        assert_eq!(
1204            redis.delete_keys_call_count(),
1205            0,
1206            "circuit-open batch delete should skip Redis"
1207        );
1208    }
1209
1210    #[test]
1211    fn escape_scan_glob_literal_escapes_redis_glob_metacharacters() {
1212        assert_eq!(escape_scan_glob_literal("plain:prefix:"), "plain:prefix:");
1213        assert_eq!(escape_scan_glob_literal("a*b"), "a\\*b");
1214        assert_eq!(escape_scan_glob_literal("a?b"), "a\\?b");
1215        assert_eq!(escape_scan_glob_literal("[ab]"), "\\[ab\\]");
1216        assert_eq!(escape_scan_glob_literal("a\\b"), "a\\\\b");
1217        // The backslash must be doubled first, or an input backslash would end up
1218        // escaping one of the escape sequences we add afterwards.
1219        assert_eq!(escape_scan_glob_literal("\\*"), "\\\\\\*");
1220    }
1221
1222    #[tokio::test]
1223    async fn invalidate_prefix_clears_local_fallback_even_when_redis_is_unavailable() {
1224        let (cache, redis) = cache_with_fake_redis(60);
1225        open_fallback_circuit(&cache);
1226        cache.set_bytes("folder:1", b"one".to_vec(), Some(60)).await;
1227        cache.set_bytes("folder:2", b"two".to_vec(), Some(60)).await;
1228        cache.set_bytes("other:1", b"keep".to_vec(), Some(60)).await;
1229
1230        cache.invalidate_prefix("folder:").await;
1231
1232        assert_eq!(cache.get_bytes("folder:1").await, None);
1233        assert_eq!(cache.get_bytes("folder:2").await, None);
1234        assert_eq!(cache.get_bytes("other:1").await, Some(b"keep".to_vec()));
1235        assert_eq!(
1236            redis.scan_call_count(),
1237            0,
1238            "circuit-open prefix invalidation should skip Redis"
1239        );
1240    }
1241
1242    #[tokio::test]
1243    async fn invalidate_prefix_deletes_matching_redis_keys_and_local_shadow() {
1244        let (cache, redis) = cache_with_fake_redis(60);
1245        redis.insert("folder:1", b"one");
1246        redis.insert("folder:2", b"two");
1247        redis.insert("other:1", b"keep");
1248        open_fallback_circuit(&cache);
1249        cache
1250            .set_bytes("folder:local", b"local".to_vec(), Some(60))
1251            .await;
1252        cache.availability.mark_success();
1253
1254        cache.invalidate_prefix("folder:").await;
1255
1256        assert!(!redis.contains_key("folder:1"));
1257        assert!(!redis.contains_key("folder:2"));
1258        assert!(redis.contains_key("other:1"));
1259        assert_eq!(cache.local.get_bytes("folder:local").await, None);
1260        assert_eq!(redis.scan_call_count(), 1);
1261        assert_eq!(redis.delete_keys_call_count(), 1);
1262    }
1263
1264    #[tokio::test]
1265    async fn invalidate_prefix_scans_and_deletes_multiple_redis_pages() {
1266        let (cache, redis) = cache_with_fake_redis(60);
1267        for index in 0..5 {
1268            redis.insert(&format!("folder:{index}"), b"value");
1269        }
1270        redis.insert("other:1", b"keep");
1271
1272        cache.invalidate_prefix("folder:").await;
1273
1274        for index in 0..5 {
1275            assert!(!redis.contains_key(&format!("folder:{index}")));
1276        }
1277        assert!(redis.contains_key("other:1"));
1278        assert_eq!(redis.scan_call_count(), 3);
1279        assert_eq!(redis.delete_keys_call_count(), 3);
1280    }
1281
1282    #[tokio::test]
1283    async fn health_check_reports_fallback_without_pinging_redis_while_circuit_is_open() {
1284        let (cache, redis) = cache_with_fake_redis(60);
1285        open_fallback_circuit(&cache);
1286
1287        let error = cache
1288            .health_check()
1289            .await
1290            .expect_err("open fallback circuit should report degraded Redis health");
1291
1292        assert!(
1293            error
1294                .to_string()
1295                .contains("redis cache is in fallback mode")
1296        );
1297        assert_eq!(redis.ping_call_count(), 0);
1298    }
1299
1300    #[tokio::test]
1301    async fn zero_ttl_set_deletes_key_instead_of_issuing_set_ex() {
1302        let (cache, redis) = cache_with_fake_redis(60);
1303        redis.insert("ephemeral", b"old");
1304
1305        cache.set_bytes("ephemeral", b"new".to_vec(), Some(0)).await;
1306
1307        assert_eq!(
1308            redis.set_call_count(),
1309            0,
1310            "zero-TTL set must not issue SETEX"
1311        );
1312        assert_eq!(redis.delete_call_count(), 1);
1313        assert!(!redis.contains_key("ephemeral"));
1314        assert!(
1315            cache.availability.unavailable_for(Instant::now()).is_none(),
1316            "zero-TTL set must not open the fallback circuit"
1317        );
1318        assert_eq!(cache.get_bytes("ephemeral").await, None);
1319        assert_eq!(
1320            redis.get_call_count(),
1321            1,
1322            "circuit stays closed, so the read reaches Redis"
1323        );
1324    }
1325
1326    #[tokio::test]
1327    async fn zero_ttl_set_if_absent_reports_absence_without_storing() {
1328        let (cache, redis) = cache_with_fake_redis(60);
1329
1330        assert!(
1331            cache
1332                .set_bytes_if_absent("nonce", b"v".to_vec(), Some(0))
1333                .await
1334        );
1335        assert_eq!(
1336            redis.set_nx_call_count(),
1337            0,
1338            "zero-TTL insert must not issue SET NX EX"
1339        );
1340        assert!(!redis.contains_key("nonce"));
1341        assert_eq!(cache.get_bytes("nonce").await, None);
1342        assert!(
1343            cache
1344                .set_bytes_if_absent("nonce", b"v2".to_vec(), Some(0))
1345                .await,
1346            "nothing is retained, so a second zero-TTL insert also succeeds"
1347        );
1348
1349        redis.insert("live", b"real");
1350        assert!(
1351            !cache
1352                .set_bytes_if_absent("live", b"v".to_vec(), Some(0))
1353                .await,
1354            "an existing live value rejects the insert"
1355        );
1356        assert_eq!(cache.get_bytes("live").await, Some(b"real".to_vec()));
1357    }
1358
1359    #[tokio::test]
1360    async fn zero_ttl_set_if_absent_with_open_circuit_uses_local_semantics() {
1361        let (cache, redis) = cache_with_fake_redis(60);
1362        open_fallback_circuit(&cache);
1363
1364        assert!(
1365            cache
1366                .set_bytes_if_absent("nonce", b"v".to_vec(), Some(0))
1367                .await
1368        );
1369        assert_eq!(
1370            redis.get_call_count(),
1371            0,
1372            "circuit-open existence check should skip Redis"
1373        );
1374        assert_eq!(cache.get_bytes("nonce").await, None);
1375        assert!(
1376            cache
1377                .set_bytes_if_absent("nonce", b"v2".to_vec(), Some(0))
1378                .await,
1379            "local zero-TTL entries expire immediately and stay insertable"
1380        );
1381    }
1382
1383    #[tokio::test]
1384    async fn zero_ttl_set_with_open_circuit_only_clears_local_shadow() {
1385        let (cache, redis) = cache_with_fake_redis(60);
1386        open_fallback_circuit(&cache);
1387        cache.set_bytes("shadow", b"local".to_vec(), Some(60)).await;
1388
1389        cache.set_bytes("shadow", b"gone".to_vec(), Some(0)).await;
1390
1391        assert_eq!(
1392            redis.delete_call_count(),
1393            0,
1394            "circuit-open zero-TTL set should skip the Redis delete"
1395        );
1396        assert_eq!(cache.local.get_bytes("shadow").await, None);
1397        assert_eq!(cache.get_bytes("shadow").await, None);
1398    }
1399
1400    #[tokio::test]
1401    async fn zero_default_ttl_treats_missing_ttl_as_immediate_expiry() {
1402        let (cache, redis) = cache_with_fake_redis(0);
1403
1404        cache.set_bytes("key", b"value".to_vec(), None).await;
1405
1406        assert_eq!(redis.set_call_count(), 0);
1407        assert_eq!(redis.delete_call_count(), 1);
1408        assert_eq!(cache.get_bytes("key").await, None);
1409    }
1410
1411    #[tokio::test]
1412    async fn command_error_falls_back_for_single_operation_without_opening_circuit() {
1413        let (cache, redis) = cache_with_fake_redis(60);
1414        redis.set_fail_command_errors(true);
1415
1416        cache
1417            .set_bytes("session", b"fallback".to_vec(), Some(60))
1418            .await;
1419
1420        assert_eq!(redis.set_call_count(), 1);
1421        assert_eq!(
1422            cache.local.get_bytes("session").await,
1423            Some(b"fallback".to_vec()),
1424            "the failed operation still falls back locally"
1425        );
1426        assert!(
1427            cache.availability.unavailable_for(Instant::now()).is_none(),
1428            "command errors must not open the fallback circuit"
1429        );
1430
1431        redis.set_fail_command_errors(false);
1432        redis.insert("session", b"redis-value");
1433        assert_eq!(
1434            cache.get_bytes("session").await,
1435            Some(b"redis-value".to_vec()),
1436            "later operations keep reaching Redis"
1437        );
1438    }
1439
1440    #[tokio::test]
1441    async fn transient_server_error_opens_fallback_circuit() {
1442        let (cache, redis) = cache_with_fake_redis(60);
1443        redis.set_fail_operations(true);
1444
1445        cache
1446            .set_bytes("session", b"value".to_vec(), Some(60))
1447            .await;
1448
1449        assert!(
1450            cache.availability.unavailable_for(Instant::now()).is_some(),
1451            "I/O errors still open the fallback circuit"
1452        );
1453    }
1454
1455    #[test]
1456    fn redis_error_indicates_unavailability_classifies_error_kinds() {
1457        use redis::{ErrorKind, ServerErrorKind};
1458
1459        fn error(kind: ErrorKind) -> redis::RedisError {
1460            redis::RedisError::from((kind, "fake error"))
1461        }
1462
1463        for kind in [
1464            ErrorKind::Io,
1465            ErrorKind::ClusterConnectionNotFound,
1466            ErrorKind::Server(ServerErrorKind::BusyLoading),
1467            ErrorKind::Server(ServerErrorKind::TryAgain),
1468            ErrorKind::Server(ServerErrorKind::ClusterDown),
1469            ErrorKind::Server(ServerErrorKind::MasterDown),
1470            ErrorKind::Server(ServerErrorKind::ReadOnly),
1471        ] {
1472            assert!(
1473                super::redis_error_indicates_unavailability(&error(kind)),
1474                "{kind:?} should indicate unavailability"
1475            );
1476        }
1477
1478        for kind in [
1479            ErrorKind::Server(ServerErrorKind::ResponseError),
1480            ErrorKind::Server(ServerErrorKind::ExecAbort),
1481            ErrorKind::Server(ServerErrorKind::NoScript),
1482            ErrorKind::Server(ServerErrorKind::Moved),
1483            ErrorKind::Server(ServerErrorKind::Ask),
1484            ErrorKind::Server(ServerErrorKind::CrossSlot),
1485            ErrorKind::Server(ServerErrorKind::NotBusy),
1486            ErrorKind::Server(ServerErrorKind::NoSub),
1487            ErrorKind::Server(ServerErrorKind::NoPerm),
1488            ErrorKind::AuthenticationFailed,
1489            ErrorKind::InvalidClientConfig,
1490            ErrorKind::UnexpectedReturnType,
1491            ErrorKind::Client,
1492            ErrorKind::Extension,
1493            ErrorKind::RESP3NotSupported,
1494            ErrorKind::Parse,
1495        ] {
1496            assert!(
1497                !super::redis_error_indicates_unavailability(&error(kind)),
1498                "{kind:?} should not indicate unavailability"
1499            );
1500        }
1501    }
1502}
aster_forge_cache/redis_cache.rs

aster_forge_cache/
redis_cache.rs
