aster_forge_cache/
lib.rs

1//! Shared cache abstractions and backend constructors for Aster services.
2//!
3//! The public API is byte-oriented so cache backends can remain object-safe and easy to wrap in
4//! `Arc<dyn CacheBackend>`. JSON convenience methods are provided as an extension trait for common
5//! application values, while concrete memory and Redis implementations live behind feature-gated
6//! modules.
7#![cfg_attr(
8    not(test),
9    deny(
10        clippy::unwrap_used,
11        clippy::unreachable,
12        clippy::expect_used,
13        clippy::panic,
14        clippy::unimplemented,
15        clippy::todo
16    )
17)]
18
19#[cfg(feature = "bloom")]
20pub mod bloom;
21#[cfg(feature = "runtime-component")]
22mod health;
23#[cfg(feature = "memory")]
24mod memory;
25#[cfg(feature = "redis")]
26mod redis_cache;
27#[cfg(feature = "memory")]
28mod reservation;
29
30use async_trait::async_trait;
31use serde::{Serialize, de::DeserializeOwned};
32use std::borrow::Cow;
33#[cfg(feature = "memory")]
34use std::sync::Arc;
35
36#[cfg(feature = "runtime-component")]
37pub use health::{
38    CACHE_COMPONENT, CACHE_HEALTH_CHECK, CACHE_HEALTH_CHECK_TIMEOUT, CacheHealthComponent,
39    cache_health_component, cache_health_options, check_cache_component,
40};
41#[cfg(feature = "memory")]
42pub use memory::MemoryCache;
43#[cfg(feature = "redis")]
44pub use redis_cache::RedisCache;
45
46/// Result type returned by cache operations.
47pub type Result<T> = std::result::Result<T, CacheError>;
48
49/// Errors returned by cache construction and health checks.
50#[derive(Debug, thiserror::Error)]
51pub enum CacheError {
52    /// Cache backend configuration is invalid or unsupported.
53    #[error("invalid cache configuration: {0}")]
54    InvalidConfiguration(String),
55    /// Redis could not be reached or initialized.
56    #[error("redis cache connection: {0}")]
57    RedisConnection(String),
58    /// Redis is temporarily unavailable and the local fallback circuit is open.
59    #[error("redis cache is in fallback mode for another {remaining_ms}ms")]
60    RedisFallbackMode {
61        /// Remaining fallback-circuit duration in milliseconds.
62        remaining_ms: u128,
63    },
64    /// Redis health check returned an error.
65    #[error("redis cache health check: {0}")]
66    RedisHealthCheck(String),
67    /// Redis health check timed out.
68    #[error("redis cache health check timed out after {timeout_ms}ms")]
69    RedisHealthCheckTimeout {
70        /// Health-check timeout in milliseconds.
71        timeout_ms: u128,
72    },
73}
74
75#[cfg(feature = "redis")]
76impl From<redis::RedisError> for CacheError {
77    fn from(value: redis::RedisError) -> Self {
78        Self::RedisConnection(value.to_string())
79    }
80}
81
82const DEFAULT_CACHE_BACKEND: &str = "memory";
83const DEFAULT_CACHE_TTL_SECS: u64 = 3600;
84
85/// Cache backend endpoint input.
86///
87/// The untagged representation preserves the existing string configuration while allowing a
88/// structured base URL plus raw credentials.
89#[derive(Clone, serde::Deserialize, serde::Serialize, PartialEq, Eq)]
90#[serde(untagged)]
91pub enum CacheEndpoint {
92    /// A complete backend URL. Existing percent-encoded Redis URLs use this mode.
93    Url(String),
94    /// A base URL without userinfo plus raw credentials.
95    Credentials {
96        /// Absolute Redis URL without username or password.
97        base_url: String,
98        /// Raw Redis ACL username.
99        #[serde(default, skip_serializing)]
100        username: Option<String>,
101        /// Raw Redis password.
102        #[serde(default, skip_serializing)]
103        password: Option<String>,
104    },
105}
106
107impl CacheEndpoint {
108    /// Creates a complete-URL endpoint.
109    pub fn url(url: impl Into<String>) -> Self {
110        Self::Url(url.into())
111    }
112
113    /// Creates a base URL plus raw credentials endpoint.
114    pub fn credentials(
115        base_url: impl Into<String>,
116        username: Option<String>,
117        password: Option<String>,
118    ) -> Self {
119        Self::Credentials {
120            base_url: base_url.into(),
121            username,
122            password,
123        }
124    }
125}
126
127impl Default for CacheEndpoint {
128    fn default() -> Self {
129        Self::Url(String::new())
130    }
131}
132
133impl From<String> for CacheEndpoint {
134    fn from(url: String) -> Self {
135        Self::Url(url)
136    }
137}
138
139impl From<&str> for CacheEndpoint {
140    fn from(url: &str) -> Self {
141        Self::Url(url.to_string())
142    }
143}
144
145impl std::fmt::Debug for CacheEndpoint {
146    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
147        match self {
148            Self::Url(_) => formatter.write_str("CacheEndpoint::Url(<redacted>)"),
149            Self::Credentials { .. } => {
150                formatter.write_str("CacheEndpoint::Credentials(<redacted>)")
151            }
152        }
153    }
154}
155
156/// Configuration used to construct a cache backend.
157#[derive(Debug, Clone, serde::Deserialize, serde::Serialize, PartialEq, Eq)]
158pub struct CacheConfig {
159    /// Backend name. Currently `memory` and `redis` are recognized.
160    #[serde(default = "CacheConfig::default_backend")]
161    pub backend: String,
162    /// Backend endpoint. Redis uses a Redis connection URL.
163    #[serde(default, alias = "redis_url")]
164    pub endpoint: CacheEndpoint,
165    /// Default time-to-live, in seconds, for entries that do not specify an explicit TTL.
166    #[serde(default = "CacheConfig::default_ttl")]
167    pub default_ttl: u64,
168}
169
170impl Default for CacheConfig {
171    fn default() -> Self {
172        Self {
173            backend: Self::default_backend(),
174            endpoint: CacheEndpoint::default(),
175            default_ttl: Self::default_ttl(),
176        }
177    }
178}
179
180/// Controls how cache construction handles a requested backend that cannot be created.
181#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
182pub enum CacheBackendFailurePolicy {
183    /// Return the construction error to the caller.
184    ReturnError,
185    /// Preserve the compatibility behavior by constructing a memory cache.
186    #[default]
187    FallbackToMemory,
188}
189
190impl CacheConfig {
191    fn default_backend() -> String {
192        DEFAULT_CACHE_BACKEND.to_string()
193    }
194
195    const fn default_ttl() -> u64 {
196        DEFAULT_CACHE_TTL_SECS
197    }
198
199    /// Returns the normalized backend name used by construction, validation, and health checks.
200    #[must_use]
201    pub fn normalized_backend(&self) -> Cow<'_, str> {
202        let backend = self.backend.trim();
203        if backend.eq_ignore_ascii_case("memory") {
204            Cow::Borrowed("memory")
205        } else if backend.eq_ignore_ascii_case("redis") {
206            Cow::Borrowed("redis")
207        } else if backend.bytes().all(|byte| !byte.is_ascii_uppercase()) {
208            Cow::Borrowed(backend)
209        } else {
210            Cow::Owned(backend.to_ascii_lowercase())
211        }
212    }
213}
214
215#[cfg(feature = "redis")]
216fn redis_backend_target_url(endpoint: &str) -> String {
217    let Some((scheme, rest)) = endpoint.split_once("://") else {
218        return "configured".to_string();
219    };
220
221    let authority = rest.split(['/', '?', '#']).next().unwrap_or_default();
222    let host = authority.rsplit('@').next().unwrap_or(authority);
223    if host.is_empty() {
224        format!("{scheme}://configured")
225    } else {
226        format!("{scheme}://{host}")
227    }
228}
229
230#[cfg(feature = "redis")]
231fn redis_backend_target(endpoint: &CacheEndpoint) -> String {
232    match endpoint {
233        CacheEndpoint::Url(url) => redis_backend_target_url(url),
234        CacheEndpoint::Credentials { base_url, .. } => redis_backend_target_url(base_url),
235    }
236}
237
238/// Object-safe cache backend trait that exposes a common byte-oriented API.
239#[async_trait]
240pub trait CacheBackend: Send + Sync {
241    /// Returns the stable backend name.
242    fn backend_name(&self) -> &'static str;
243    /// Performs a lightweight backend health check.
244    async fn health_check(&self) -> Result<()>;
245    /// Reads a raw byte value by key.
246    async fn get_bytes(&self, key: &str) -> Option<Vec<u8>>;
247    /// Atomically reads and removes a raw byte value by key when supported.
248    async fn take_bytes(&self, key: &str) -> Option<Vec<u8>>;
249    /// Writes a raw byte value by key with an optional TTL in seconds.
250    ///
251    /// `None` uses the backend default TTL. A TTL of `0` expires immediately, so the
252    /// observable result matches deleting the key (Redis rejects `SETEX 0`; backends
253    /// normalize the contract instead of issuing an invalid command).
254    async fn set_bytes(&self, key: &str, value: Vec<u8>, ttl_secs: Option<u64>);
255    /// Writes a raw byte value only when the key is absent.
256    ///
257    /// With a TTL of `0` the value expires immediately, so the call reports whether a
258    /// live value existed and retains nothing either way.
259    async fn set_bytes_if_absent(&self, key: &str, value: Vec<u8>, ttl_secs: Option<u64>) -> bool;
260    /// Removes a key from the cache.
261    async fn delete(&self, key: &str);
262    /// Removes multiple keys from the cache.
263    async fn delete_many(&self, keys: &[String]) {
264        for key in keys {
265            self.delete(key).await;
266        }
267    }
268    /// Invalidates every key with the given prefix.
269    async fn invalidate_prefix(&self, prefix: &str);
270}
271
272/// Convenience methods for JSON serialization and deserialization.
273pub trait CacheExt {
274    /// Reads and deserializes a JSON value from the cache.
275    fn get<T: DeserializeOwned + Send>(
276        &self,
277        key: &str,
278    ) -> impl std::future::Future<Output = Option<T>> + Send;
279
280    /// Serializes and writes a JSON value to the cache.
281    fn set<T: Serialize + Send + Sync>(
282        &self,
283        key: &str,
284        value: &T,
285        ttl_secs: Option<u64>,
286    ) -> impl std::future::Future<Output = ()> + Send;
287
288    /// Atomically reads, removes, and deserializes a JSON value from the cache.
289    fn take<T: DeserializeOwned + Send>(
290        &self,
291        key: &str,
292    ) -> impl std::future::Future<Output = Option<T>> + Send;
293}
294
295impl CacheExt for dyn CacheBackend {
296    async fn get<T: DeserializeOwned + Send>(&self, key: &str) -> Option<T> {
297        let bytes = self.get_bytes(key).await?;
298        serde_json::from_slice(&bytes).ok()
299    }
300
301    async fn set<T: Serialize + Send + Sync>(&self, key: &str, value: &T, ttl_secs: Option<u64>) {
302        if let Ok(bytes) = serde_json::to_vec(value) {
303            self.set_bytes(key, bytes, ttl_secs).await;
304        }
305    }
306
307    async fn take<T: DeserializeOwned + Send>(&self, key: &str) -> Option<T> {
308        let bytes = self.take_bytes(key).await?;
309        serde_json::from_slice(&bytes).ok()
310    }
311}
312
313/// Creates a cache backend with an explicit construction-failure policy.
314///
315/// # Errors
316///
317/// Returns [`CacheError`] when the requested backend is unsupported or Redis initialization fails
318/// and [`CacheBackendFailurePolicy::ReturnError`] is selected.
319#[cfg(feature = "memory")]
320#[cfg_attr(
321    not(feature = "redis"),
322    expect(
323        clippy::unused_async,
324        reason = "The Redis feature awaits connection setup; keep one factory API across feature sets."
325    )
326)]
327pub async fn create_cache_with_policy(
328    config: &CacheConfig,
329    failure_policy: CacheBackendFailurePolicy,
330) -> Result<Arc<dyn CacheBackend>> {
331    match config.normalized_backend().as_ref() {
332        #[cfg(feature = "redis")]
333        "redis" => {
334            match redis_cache::RedisCache::from_endpoint(&config.endpoint, config.default_ttl).await
335            {
336                Ok(cache) => {
337                    tracing::info!(
338                        target = %redis_backend_target(&config.endpoint),
339                        "cache backend: redis"
340                    );
341                    Ok(Arc::new(cache))
342                }
343                Err(error) => handle_cache_backend_failure(config, failure_policy, error),
344            }
345        }
346        "memory" => {
347            tracing::info!("cache backend: memory (ttl={}s)", config.default_ttl);
348            Ok(create_memory_cache(config.default_ttl))
349        }
350        backend => handle_cache_backend_failure(
351            config,
352            failure_policy,
353            CacheError::InvalidConfiguration(format!("unsupported cache backend '{backend}'")),
354        ),
355    }
356}
357
358#[cfg(feature = "memory")]
359fn handle_cache_backend_failure(
360    config: &CacheConfig,
361    failure_policy: CacheBackendFailurePolicy,
362    error: CacheError,
363) -> Result<Arc<dyn CacheBackend>> {
364    match failure_policy {
365        CacheBackendFailurePolicy::ReturnError => Err(error),
366        CacheBackendFailurePolicy::FallbackToMemory => {
367            tracing::warn!(
368                error = %error,
369                "cache backend construction failed; falling back to memory cache"
370            );
371            Ok(create_memory_cache(config.default_ttl))
372        }
373    }
374}
375
376/// Creates a cache backend using the historical memory-fallback behavior.
377///
378/// Runtime assembly code should prefer [`create_cache_with_policy`] so the availability contract is
379/// explicit. This compatibility entry point keeps existing callers source-compatible.
380#[cfg(feature = "memory")]
381pub async fn create_cache(config: &CacheConfig) -> Arc<dyn CacheBackend> {
382    match create_cache_with_policy(config, CacheBackendFailurePolicy::FallbackToMemory).await {
383        Ok(cache) => cache,
384        Err(error) => {
385            tracing::warn!(
386                error = %error,
387                "cache compatibility constructor encountered an unexpected error; using memory cache"
388            );
389            create_memory_cache(config.default_ttl)
390        }
391    }
392}
393
394#[cfg(feature = "memory")]
395fn create_memory_cache(default_ttl: u64) -> Arc<dyn CacheBackend> {
396    Arc::new(memory::MemoryCache::new(default_ttl))
397}
398
399#[cfg(test)]
400mod tests {
401    use super::{CacheBackendFailurePolicy, CacheConfig, CacheEndpoint, CacheError};
402
403    #[test]
404    fn cache_config_default_uses_memory_backend() {
405        let config = CacheConfig::default();
406
407        assert_eq!(config.backend, "memory");
408        assert_eq!(config.endpoint, CacheEndpoint::default());
409        assert_eq!(config.default_ttl, 3600);
410    }
411
412    #[test]
413    fn cache_config_deserializes_missing_fields_with_defaults() {
414        let config: CacheConfig =
415            serde_json::from_str("{}").expect("empty cache config should use field defaults");
416
417        assert_eq!(config, CacheConfig::default());
418    }
419
420    #[test]
421    fn cache_config_deserializes_endpoint_field() {
422        let config: CacheConfig = serde_json::from_str(
423            r#"{"backend":"redis","endpoint":"redis://127.0.0.1/","default_ttl":30}"#,
424        )
425        .expect("cache config should accept the endpoint field");
426
427        assert_eq!(config.backend, "redis");
428        assert_eq!(config.endpoint, CacheEndpoint::url("redis://127.0.0.1/"));
429        assert_eq!(config.default_ttl, 30);
430    }
431
432    #[test]
433    fn cache_config_accepts_legacy_redis_url_alias() {
434        let config: CacheConfig = serde_json::from_str(
435            r#"{"backend":"redis","redis_url":"redis://127.0.0.1/","default_ttl":30}"#,
436        )
437        .expect("cache config should accept legacy redis_url config files");
438
439        assert_eq!(config.backend, "redis");
440        assert_eq!(config.endpoint, CacheEndpoint::url("redis://127.0.0.1/"));
441        assert_eq!(config.default_ttl, 30);
442    }
443
444    #[test]
445    fn cache_backend_normalization_trims_and_folds_ascii_case() {
446        for (backend, expected) in [
447            (" memory ", "memory"),
448            (" ReDiS ", "redis"),
449            (" CUSTOM-BACKEND ", "custom-backend"),
450            (" \n\t", ""),
451        ] {
452            let config = CacheConfig {
453                backend: backend.to_string(),
454                ..CacheConfig::default()
455            };
456            assert_eq!(config.normalized_backend(), expected);
457        }
458    }
459
460    #[cfg(feature = "memory")]
461    #[tokio::test]
462    async fn create_cache_uses_normalized_memory_backend() {
463        let cache = super::create_cache(&CacheConfig {
464            backend: " MeMoRy ".to_string(),
465            ..CacheConfig::default()
466        })
467        .await;
468
469        assert_eq!(cache.backend_name(), "memory");
470    }
471
472    #[cfg(feature = "memory")]
473    #[tokio::test]
474    async fn create_cache_uses_memory_for_unknown_backend() {
475        let cache = super::create_cache(&CacheConfig {
476            backend: "unknown".to_string(),
477            endpoint: "redis://127.0.0.1/".into(),
478            default_ttl: 5,
479        })
480        .await;
481
482        assert_eq!(cache.backend_name(), "memory");
483        cache.health_check().await.expect("memory cache is healthy");
484    }
485
486    #[cfg(feature = "memory")]
487    #[tokio::test]
488    async fn explicit_error_policy_rejects_unknown_backend() {
489        let Err(error) = super::create_cache_with_policy(
490            &CacheConfig {
491                backend: "unknown".to_string(),
492                ..CacheConfig::default()
493            },
494            CacheBackendFailurePolicy::ReturnError,
495        )
496        .await
497        else {
498            panic!("unknown backend should be returned to explicit callers");
499        };
500
501        assert!(
502            error
503                .to_string()
504                .contains("unsupported cache backend 'unknown'")
505        );
506    }
507
508    #[cfg(feature = "redis")]
509    #[tokio::test]
510    async fn explicit_error_policy_returns_redis_construction_error() {
511        let Err(error) = super::create_cache_with_policy(
512            &CacheConfig {
513                backend: "redis".to_string(),
514                endpoint: "not a redis url".into(),
515                default_ttl: 60,
516            },
517            CacheBackendFailurePolicy::ReturnError,
518        )
519        .await
520        else {
521            panic!("invalid Redis endpoint should be returned to explicit callers");
522        };
523
524        assert!(error.to_string().contains("redis cache connection"));
525    }
526
527    #[cfg(feature = "redis")]
528    #[test]
529    fn redis_backend_target_strips_credentials() {
530        assert_eq!(
531            super::redis_backend_target(&CacheEndpoint::url(
532                "redis://user:secret@example.com:6379/0",
533            )),
534            "redis://example.com:6379"
535        );
536    }
537
538    #[cfg(feature = "redis")]
539    #[test]
540    fn redis_backend_target_keeps_host_without_credentials() {
541        assert_eq!(
542            super::redis_backend_target(&CacheEndpoint::url("rediss://cache.internal:6380/1",)),
543            "rediss://cache.internal:6380"
544        );
545    }
546
547    #[cfg(feature = "redis")]
548    #[test]
549    fn redis_backend_target_handles_malformed_or_empty_hosts() {
550        assert_eq!(
551            super::redis_backend_target(&CacheEndpoint::url("not-a-url")),
552            "configured"
553        );
554        assert_eq!(
555            super::redis_backend_target(&CacheEndpoint::url("redis:///0")),
556            "redis://configured"
557        );
558    }
559
560    #[test]
561    fn cache_config_deserializes_structured_credentials_and_redacts_debug() {
562        let raw_username = "cache-user@example.com";
563        let raw_password = "cache#[]{}^+=*@:/?%secret";
564        let config: CacheConfig = serde_json::from_str(&format!(
565            r#"{{"backend":"redis","endpoint":{{"base_url":"redis://cache.example/0","username":"{raw_username}","password":"{raw_password}"}}}}"#,
566        ))
567        .unwrap();
568
569        assert_eq!(
570            config.endpoint,
571            CacheEndpoint::credentials(
572                "redis://cache.example/0",
573                Some(raw_username.to_string()),
574                Some(raw_password.to_string()),
575            )
576        );
577        let debug = format!("{config:?}");
578        assert!(!debug.contains(raw_username));
579        assert!(!debug.contains(raw_password));
580        assert!(debug.contains("CacheEndpoint::Credentials(<redacted>)"));
581
582        let serialized = serde_json::to_string(&config).unwrap();
583        assert!(serialized.contains("redis://cache.example/0"));
584        assert!(!serialized.contains(raw_username));
585        assert!(!serialized.contains(raw_password));
586        assert!(!serialized.contains("cache%23%5B%5D"));
587    }
588
589    #[test]
590    fn cache_error_display_messages_are_stable() {
591        assert_eq!(
592            CacheError::InvalidConfiguration("bad backend".to_string()).to_string(),
593            "invalid cache configuration: bad backend"
594        );
595        assert_eq!(
596            CacheError::RedisConnection("refused".to_string()).to_string(),
597            "redis cache connection: refused"
598        );
599        assert_eq!(
600            CacheError::RedisFallbackMode { remaining_ms: 25 }.to_string(),
601            "redis cache is in fallback mode for another 25ms"
602        );
603        assert_eq!(
604            CacheError::RedisHealthCheck("PONG missing".to_string()).to_string(),
605            "redis cache health check: PONG missing"
606        );
607        assert_eq!(
608            CacheError::RedisHealthCheckTimeout { timeout_ms: 250 }.to_string(),
609            "redis cache health check timed out after 250ms"
610        );
611    }
612}
aster_forge_cache/lib.rs

aster_forge_cache/
lib.rs
