cachekit/backend/
redis.rs

1use std::collections::HashMap;
2use std::time::Duration;
3
4use async_trait::async_trait;
5use fred::interfaces::LuaInterface;
6use fred::prelude::*;
7use fred::types::ConnectHandle;
8
9use crate::backend::{nonzero_timeout, Backend, HealthStatus, LockableBackend, TtlInspectable};
10use crate::error::{BackendError, BackendErrorKind};
11
12// ── Error mapping ─────────────────────────────────────────────────────────────
13
14/// Sanitize Redis error messages to avoid leaking connection URL passwords.
15fn sanitize_redis_message(msg: &str) -> String {
16    // Redis URLs may contain passwords: redis://:password@host:6379
17    // Strip anything between :// and @ to remove embedded credentials.
18    if let Some(proto_end) = msg.find("://") {
19        if let Some(at_pos) = msg[proto_end..].find('@') {
20            let mut sanitized = String::with_capacity(msg.len());
21            sanitized.push_str(&msg[..proto_end + 3]);
22            sanitized.push_str("[REDACTED]");
23            sanitized.push_str(&msg[proto_end + at_pos..]);
24            return sanitized;
25        }
26    }
27    msg.to_string()
28}
29
30fn redis_err(e: RedisError) -> BackendError {
31    let kind = match e.kind() {
32        RedisErrorKind::Auth => BackendErrorKind::Authentication,
33        RedisErrorKind::IO => BackendErrorKind::Transient,
34        RedisErrorKind::Timeout => BackendErrorKind::Timeout,
35        RedisErrorKind::Canceled => BackendErrorKind::Transient,
36        _ => BackendErrorKind::Permanent,
37    };
38    BackendError {
39        kind,
40        message: sanitize_redis_message(&e.to_string()),
41        source: Some(Box::new(e)),
42    }
43}
44
45// ── RedisBackend ──────────────────────────────────────────────────────────────
46
47/// Redis backend powered by the `fred` client.
48///
49/// Call [`connect`](RedisBackend::connect) before issuing any cache operations.
50pub struct RedisBackend {
51    client: RedisClient,
52}
53
54impl std::fmt::Debug for RedisBackend {
55    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
56        f.debug_struct("RedisBackend").finish_non_exhaustive()
57    }
58}
59
60impl RedisBackend {
61    /// Start building a [`RedisBackend`].
62    pub fn builder() -> RedisBackendBuilder {
63        RedisBackendBuilder::default()
64    }
65
66    /// Connect to Redis and wait until the connection is established.
67    ///
68    /// This must be called before any cache operations.
69    ///
70    /// The returned `ConnectHandle` is fred's handle to the spawned task that
71    /// drives the connection (a tokio `JoinHandle`). Dropping it detaches the
72    /// task without stopping it: the connection, and any auto-reconnect,
73    /// keeps running for as long as the backend is alive. The intent presets
74    /// drop it for that reason. Keep it only to observe how the connection
75    /// ended: the task resolves when the connection closes for good (on a
76    /// fail-fast backend, the first dropped connection), with the error that
77    /// closed it, if any. Do not `abort` it while the backend is still in use.
78    pub async fn connect(&self) -> Result<ConnectHandle, BackendError> {
79        self.client.init().await.map_err(redis_err)
80    }
81
82    /// The fred client, so tests can read the options `build` gave it.
83    #[cfg(test)]
84    pub(crate) fn client(&self) -> &RedisClient {
85        &self.client
86    }
87}
88
89// ── Backend impl ──────────────────────────────────────────────────────────────
90
91#[cfg(not(target_arch = "wasm32"))]
92#[cfg_attr(not(feature = "unsync"), async_trait)]
93#[cfg_attr(feature = "unsync", async_trait(?Send))]
94impl Backend for RedisBackend {
95    async fn get(&self, key: &str) -> Result<Option<Vec<u8>>, BackendError> {
96        let result: Option<bytes::Bytes> = self.client.get(key).await.map_err(redis_err)?;
97        Ok(result.map(|b| b.to_vec()))
98    }
99
100    async fn set(
101        &self,
102        key: &str,
103        value: Vec<u8>,
104        ttl: Option<Duration>,
105    ) -> Result<(), BackendError> {
106        let expiration = ttl.map(|d| {
107            let secs = i64::try_from(d.as_secs().max(1)).unwrap_or(i64::MAX);
108            Expiration::EX(secs)
109        });
110        self.client
111            .set::<(), _, _>(key, value.as_slice(), expiration, None, false)
112            .await
113            .map_err(redis_err)
114    }
115
116    async fn delete(&self, key: &str) -> Result<bool, BackendError> {
117        let removed: i64 = self.client.del(key).await.map_err(redis_err)?;
118        Ok(removed > 0)
119    }
120
121    async fn exists(&self, key: &str) -> Result<bool, BackendError> {
122        let count: i64 = self.client.exists(key).await.map_err(redis_err)?;
123        Ok(count > 0)
124    }
125
126    async fn health(&self) -> Result<HealthStatus, BackendError> {
127        let start = std::time::Instant::now();
128        let _pong: String = self.client.ping().await.map_err(redis_err)?;
129        let latency = start.elapsed();
130
131        let mut details = HashMap::new();
132        details.insert("latency_ms".to_string(), latency.as_millis().to_string());
133        Ok(HealthStatus {
134            is_healthy: true,
135            latency_ms: latency.as_secs_f64() * 1000.0,
136            backend_type: "redis".to_string(),
137            details,
138        })
139    }
140
141    fn as_lockable(&self) -> Option<&dyn LockableBackend> {
142        Some(self)
143    }
144}
145
146// ── TtlInspectable impl ───────────────────────────────────────────────────────
147
148#[cfg(not(target_arch = "wasm32"))]
149#[cfg_attr(not(feature = "unsync"), async_trait)]
150#[cfg_attr(feature = "unsync", async_trait(?Send))]
151impl TtlInspectable for RedisBackend {
152    async fn ttl(&self, key: &str) -> Result<Option<Duration>, BackendError> {
153        // Redis TTL return values:
154        //   -2 → key does not exist
155        //   -1 → key exists but has no TTL
156        //   N  → remaining seconds
157        let secs: i64 = self.client.ttl(key).await.map_err(redis_err)?;
158        match secs {
159            ..0 => Ok(None),
160            n => Ok(Some(Duration::from_secs(n.unsigned_abs()))),
161        }
162    }
163}
164
165// ── LockableBackend impl ──────────────────────────────────────────────────────
166
167/// Callers pass the bare cache key; the backend derives the lock key on the
168/// wire. `<key>:lock` matches cachekit-py's Redis lock namespace so py and rs
169/// workloads contend on the same lock.
170fn lock_key(key: &str) -> String {
171    format!("{key}:lock")
172}
173
174/// Compare-and-delete: only the holder of the lock_id token may release.
175/// Atomic via Lua — a GET-then-DEL from the client would race with expiry
176/// and delete a lock some other client has since acquired.
177const RELEASE_LOCK_SCRIPT: &str = r#"if redis.call("GET", KEYS[1]) == ARGV[1] then return redis.call("DEL", KEYS[1]) else return 0 end"#;
178
179#[cfg(not(target_arch = "wasm32"))]
180#[cfg_attr(not(feature = "unsync"), async_trait)]
181#[cfg_attr(feature = "unsync", async_trait(?Send))]
182impl LockableBackend for RedisBackend {
183    async fn acquire_lock(
184        &self,
185        key: &str,
186        timeout_ms: u64,
187    ) -> Result<Option<String>, BackendError> {
188        let lock_id = uuid::Uuid::new_v4().to_string();
189        // PX must be >= 1 (Redis rejects 0); clamp like Backend::set does for EX.
190        let px = i64::try_from(timeout_ms.max(1)).unwrap_or(i64::MAX);
191
192        // SET NX PX: returns OK when the lock was free, nil when contested.
193        let result: Option<String> = self
194            .client
195            .set(
196                lock_key(key),
197                lock_id.as_str(),
198                Some(Expiration::PX(px)),
199                Some(SetOptions::NX),
200                false,
201            )
202            .await
203            .map_err(redis_err)?;
204
205        Ok(result.map(|_| lock_id))
206    }
207
208    async fn release_lock(&self, key: &str, lock_id: &str) -> Result<bool, BackendError> {
209        let deleted: i64 = self
210            .client
211            .eval(RELEASE_LOCK_SCRIPT, vec![lock_key(key)], vec![lock_id])
212            .await
213            .map_err(redis_err)?;
214        Ok(deleted == 1)
215    }
216}
217
218// ── Builder ───────────────────────────────────────────────────────────────────
219
220/// Default per-command deadline ([`RedisBackendBuilder::command_timeout`]).
221/// fred's default is 0 (no timeout), so a stalled or black-holed server would
222/// hold every op until TCP gives up and the circuit breaker could never open.
223/// 5 s matches cachekit-py and cachekit-ts.
224pub(crate) const COMMAND_TIMEOUT: Duration = Duration::from_secs(5);
225
226/// fred client options for [`RedisBackendBuilder::build`].
227///
228/// The unresponsive-connection check is enabled only with a reconnect policy:
229/// fred closes a connection it deems unresponsive, and without a policy it is
230/// never re-established, so one stall past the threshold would leave the
231/// client failing permanently. Its threshold is the command timeout, so a
232/// connection that cannot answer within the budget is replaced.
233pub(crate) fn fred_options(
234    reconnects: bool,
235    command_timeout: Duration,
236) -> (PerformanceConfig, ConnectionConfig) {
237    let perf = PerformanceConfig {
238        default_command_timeout: command_timeout,
239        ..Default::default()
240    };
241    let mut connection = ConnectionConfig::default();
242    if reconnects {
243        connection.unresponsive.max_timeout = Some(command_timeout);
244    }
245    (perf, connection)
246}
247
248/// Builder for [`RedisBackend`].
249#[derive(Default)]
250#[must_use]
251pub struct RedisBackendBuilder {
252    url: Option<String>,
253    reconnect: Option<ReconnectPolicy>,
254    command_timeout: Option<Duration>,
255}
256
257impl RedisBackendBuilder {
258    /// Set the Redis connection URL (required).
259    ///
260    /// Accepts standard Redis URL formats, e.g. `redis://localhost:6379` or
261    /// `redis://:password@host:6379/0`.
262    pub fn url(mut self, url: impl Into<String>) -> Self {
263        self.url = Some(url.into());
264        self
265    }
266
267    /// Automatically re-establish dropped connections with exponential
268    /// backoff (100 ms → 30 s, retrying indefinitely). Off by default.
269    ///
270    /// Also turns on fred's unresponsive-connection check: a connection with
271    /// a reply pending for longer than the
272    /// [command timeout](Self::command_timeout) is closed and re-established
273    /// (fred checks every 2 s). Without a reconnect policy that check stays
274    /// off, because a closed connection would never come back.
275    ///
276    /// Initial connections still fail fast — fred's `fail_fast` default stays
277    /// on, so the policy only governs reconnecting a connection that was
278    /// previously established. This is the policy `CacheKit::production` and
279    /// `CacheKit::secure` use. Call it when you build a backend yourself, for
280    /// example one backend shared by several clients through
281    /// [`CacheKitBuilder::backend`](crate::CacheKitBuilder::backend), so it
282    /// recovers after a Redis restart instead of failing every operation.
283    ///
284    /// ```no_run
285    /// # async fn example() -> Result<(), cachekit::CachekitError> {
286    /// use cachekit::backend::redis::RedisBackend;
287    ///
288    /// let backend = RedisBackend::builder()
289    ///     .url("redis://localhost:6379")
290    ///     .auto_reconnect()
291    ///     .build()?;
292    /// let _connection = backend.connect().await?;
293    /// # Ok(())
294    /// # }
295    /// ```
296    pub fn auto_reconnect(mut self) -> Self {
297        self.reconnect = Some(ReconnectPolicy::new_exponential(0, 100, 30_000, 2));
298        self
299    }
300
301    /// Deadline for each Redis command (default 5 s). A command past it fails
302    /// with `BackendErrorKind::Timeout`, and may still run on the server.
303    ///
304    /// The budget is per attempt. When the reliability stack wraps the
305    /// backend (`CacheKit::production`, `CacheKit::secure`, or
306    /// `CacheKitBuilder::reliability`),
307    /// each retry gets a fresh budget, so with the default 3 attempts one
308    /// operation can take about three budgets plus backoff before it errors.
309    /// The stack sets no overall deadline for Redis. Each operation that ends
310    /// in a timeout counts as one failure toward opening the circuit breaker,
311    /// which a `tokio::time::timeout` around the call does not: a cancelled
312    /// call records no outcome. With [`auto_reconnect`](Self::auto_reconnect)
313    /// the budget is also the unresponsive-connection threshold.
314    ///
315    /// `build` rejects `Duration::ZERO` with `CachekitError::Config`; fred
316    /// reads zero as "no timeout".
317    pub fn command_timeout(mut self, timeout: Duration) -> Self {
318        self.command_timeout = Some(timeout);
319        self
320    }
321
322    /// Consume the builder and construct a [`RedisBackend`].
323    ///
324    /// # Errors
325    ///
326    /// Returns an error if `url` was not provided or is not a valid Redis URL,
327    /// or if the command timeout is zero. The connection itself is lazy — call [`RedisBackend::connect`] to establish it.
328    pub fn build(self) -> Result<RedisBackend, crate::error::CachekitError> {
329        use crate::error::CachekitError;
330
331        let url = self
332            .url
333            .filter(|u| !u.is_empty())
334            .ok_or_else(|| CachekitError::Config("url is required".to_string()))?;
335
336        let config = RedisConfig::from_url(&url).map_err(|e| {
337            CachekitError::Config(format!(
338                "invalid Redis URL: {}",
339                sanitize_redis_message(&e.to_string())
340            ))
341        })?;
342
343        let command_timeout =
344            nonzero_timeout("command_timeout", self.command_timeout, COMMAND_TIMEOUT)?;
345
346        let (perf, connection) = fred_options(self.reconnect.is_some(), command_timeout);
347        let client = RedisClient::new(config, Some(perf), Some(connection), self.reconnect);
348        Ok(RedisBackend { client })
349    }
350}
351
352#[cfg(test)]
353#[allow(clippy::expect_used)] // test-only: a builder failure should panic loudly
354mod tests {
355    use super::*;
356
357    /// Compile-time proof that RedisBackend implements LockableBackend.
358    fn _assert_lockable(_b: &dyn LockableBackend) {}
359
360    #[test]
361    fn redis_is_lockable() {
362        fn _check(backend: &RedisBackend) {
363            _assert_lockable(backend);
364        }
365    }
366
367    #[test]
368    fn lock_key_derives_py_compatible_namespace() {
369        // Bare-key contract: callers never build ":lock" themselves; the
370        // backend appends it, byte-identical to cachekit-py's Redis lock.
371        assert_eq!(
372            lock_key("ns:app:func:m.f:args:abc:v1"),
373            "ns:app:func:m.f:args:abc:v1:lock"
374        );
375    }
376
377    #[test]
378    fn command_timeout_is_five_seconds_on_every_build() {
379        assert_eq!(COMMAND_TIMEOUT, Duration::from_secs(5));
380        for reconnects in [false, true] {
381            let (perf, _) = fred_options(reconnects, COMMAND_TIMEOUT);
382            assert_eq!(perf.default_command_timeout, COMMAND_TIMEOUT);
383        }
384    }
385
386    #[test]
387    fn unresponsive_recycling_only_with_reconnect_policy() {
388        let (_, connection) = fred_options(false, COMMAND_TIMEOUT);
389        assert_eq!(connection.unresponsive.max_timeout, None);
390        let (_, connection) = fred_options(true, COMMAND_TIMEOUT);
391        assert_eq!(connection.unresponsive.max_timeout, Some(COMMAND_TIMEOUT));
392    }
393
394    const URL: &str = "redis://localhost:6379";
395
396    #[test]
397    fn auto_reconnect_is_off_by_default() {
398        let backend = RedisBackend::builder().url(URL).build().expect("builds");
399        assert_eq!(backend.client().client_reconnect_policy(), None);
400        assert_eq!(
401            backend
402                .client()
403                .connection_config()
404                .unresponsive
405                .max_timeout,
406            None
407        );
408    }
409
410    #[test]
411    fn auto_reconnect_sets_the_exponential_policy_and_unresponsive_check() {
412        let backend = RedisBackend::builder()
413            .url(URL)
414            .auto_reconnect()
415            .build()
416            .expect("builds");
417        assert_eq!(
418            backend.client().client_reconnect_policy(),
419            Some(ReconnectPolicy::new_exponential(0, 100, 30_000, 2))
420        );
421        assert_eq!(
422            backend
423                .client()
424                .connection_config()
425                .unresponsive
426                .max_timeout,
427            Some(COMMAND_TIMEOUT)
428        );
429    }
430
431    #[test]
432    fn configured_command_timeout_reaches_fred() {
433        let budget = Duration::from_millis(250);
434        let backend = RedisBackend::builder()
435            .url(URL)
436            .command_timeout(budget)
437            .build()
438            .expect("builds");
439        assert_eq!(
440            backend.client().perf_config().default_command_timeout,
441            budget
442        );
443        assert_eq!(
444            backend
445                .client()
446                .connection_config()
447                .unresponsive
448                .max_timeout,
449            None
450        );
451
452        let backend = RedisBackend::builder()
453            .url(URL)
454            .command_timeout(budget)
455            .auto_reconnect()
456            .build()
457            .expect("builds");
458        assert_eq!(
459            backend.client().perf_config().default_command_timeout,
460            budget
461        );
462        assert_eq!(
463            backend
464                .client()
465                .connection_config()
466                .unresponsive
467                .max_timeout,
468            Some(budget)
469        );
470    }
471
472    #[test]
473    fn default_command_timeout_reaches_fred() {
474        let backend = RedisBackend::builder().url(URL).build().expect("builds");
475        assert_eq!(
476            backend.client().perf_config().default_command_timeout,
477            COMMAND_TIMEOUT
478        );
479    }
480
481    #[test]
482    fn zero_command_timeout_is_a_config_error() {
483        let err = RedisBackend::builder()
484            .url(URL)
485            .command_timeout(Duration::ZERO)
486            .build()
487            .expect_err("zero is rejected");
488        assert!(
489            matches!(&err, crate::error::CachekitError::Config(m) if m.contains("command_timeout")),
490            "{err:?}"
491        );
492    }
493}
cachekit/backend/redis.rs

cachekit/backend/
redis.rs
