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

aster_forge_cache/
redis_cache.rs
