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")]
320pub async fn create_cache_with_policy(
321    config: &CacheConfig,
322    failure_policy: CacheBackendFailurePolicy,
323) -> Result<Arc<dyn CacheBackend>> {
324    match config.normalized_backend().as_ref() {
325        #[cfg(feature = "redis")]
326        "redis" => {
327            match redis_cache::RedisCache::from_endpoint(&config.endpoint, config.default_ttl).await
328            {
329                Ok(cache) => {
330                    tracing::info!(
331                        target = %redis_backend_target(&config.endpoint),
332                        "cache backend: redis"
333                    );
334                    Ok(Arc::new(cache))
335                }
336                Err(error) => handle_cache_backend_failure(config, failure_policy, error),
337            }
338        }
339        "memory" => {
340            tracing::info!("cache backend: memory (ttl={}s)", config.default_ttl);
341            Ok(create_memory_cache(config.default_ttl))
342        }
343        backend => handle_cache_backend_failure(
344            config,
345            failure_policy,
346            CacheError::InvalidConfiguration(format!("unsupported cache backend '{backend}'")),
347        ),
348    }
349}
350
351#[cfg(feature = "memory")]
352fn handle_cache_backend_failure(
353    config: &CacheConfig,
354    failure_policy: CacheBackendFailurePolicy,
355    error: CacheError,
356) -> Result<Arc<dyn CacheBackend>> {
357    match failure_policy {
358        CacheBackendFailurePolicy::ReturnError => Err(error),
359        CacheBackendFailurePolicy::FallbackToMemory => {
360            tracing::warn!(
361                error = %error,
362                "cache backend construction failed; falling back to memory cache"
363            );
364            Ok(create_memory_cache(config.default_ttl))
365        }
366    }
367}
368
369/// Creates a cache backend using the historical memory-fallback behavior.
370///
371/// Runtime assembly code should prefer [`create_cache_with_policy`] so the availability contract is
372/// explicit. This compatibility entry point keeps existing callers source-compatible.
373#[cfg(feature = "memory")]
374pub async fn create_cache(config: &CacheConfig) -> Arc<dyn CacheBackend> {
375    match create_cache_with_policy(config, CacheBackendFailurePolicy::FallbackToMemory).await {
376        Ok(cache) => cache,
377        Err(error) => {
378            tracing::warn!(
379                error = %error,
380                "cache compatibility constructor encountered an unexpected error; using memory cache"
381            );
382            create_memory_cache(config.default_ttl)
383        }
384    }
385}
386
387#[cfg(feature = "memory")]
388fn create_memory_cache(default_ttl: u64) -> Arc<dyn CacheBackend> {
389    Arc::new(memory::MemoryCache::new(default_ttl))
390}
391
392#[cfg(test)]
393mod tests {
394    use super::{CacheBackendFailurePolicy, CacheConfig, CacheEndpoint, CacheError};
395
396    #[test]
397    fn cache_config_default_uses_memory_backend() {
398        let config = CacheConfig::default();
399
400        assert_eq!(config.backend, "memory");
401        assert_eq!(config.endpoint, CacheEndpoint::default());
402        assert_eq!(config.default_ttl, 3600);
403    }
404
405    #[test]
406    fn cache_config_deserializes_missing_fields_with_defaults() {
407        let config: CacheConfig =
408            serde_json::from_str("{}").expect("empty cache config should use field defaults");
409
410        assert_eq!(config, CacheConfig::default());
411    }
412
413    #[test]
414    fn cache_config_deserializes_endpoint_field() {
415        let config: CacheConfig = serde_json::from_str(
416            r#"{"backend":"redis","endpoint":"redis://127.0.0.1/","default_ttl":30}"#,
417        )
418        .expect("cache config should accept the endpoint field");
419
420        assert_eq!(config.backend, "redis");
421        assert_eq!(config.endpoint, CacheEndpoint::url("redis://127.0.0.1/"));
422        assert_eq!(config.default_ttl, 30);
423    }
424
425    #[test]
426    fn cache_config_accepts_legacy_redis_url_alias() {
427        let config: CacheConfig = serde_json::from_str(
428            r#"{"backend":"redis","redis_url":"redis://127.0.0.1/","default_ttl":30}"#,
429        )
430        .expect("cache config should accept legacy redis_url config files");
431
432        assert_eq!(config.backend, "redis");
433        assert_eq!(config.endpoint, CacheEndpoint::url("redis://127.0.0.1/"));
434        assert_eq!(config.default_ttl, 30);
435    }
436
437    #[test]
438    fn cache_backend_normalization_trims_and_folds_ascii_case() {
439        for (backend, expected) in [
440            (" memory ", "memory"),
441            (" ReDiS ", "redis"),
442            (" CUSTOM-BACKEND ", "custom-backend"),
443            (" \n\t", ""),
444        ] {
445            let config = CacheConfig {
446                backend: backend.to_string(),
447                ..CacheConfig::default()
448            };
449            assert_eq!(config.normalized_backend(), expected);
450        }
451    }
452
453    #[cfg(feature = "memory")]
454    #[tokio::test]
455    async fn create_cache_uses_normalized_memory_backend() {
456        let cache = super::create_cache(&CacheConfig {
457            backend: " MeMoRy ".to_string(),
458            ..CacheConfig::default()
459        })
460        .await;
461
462        assert_eq!(cache.backend_name(), "memory");
463    }
464
465    #[cfg(feature = "memory")]
466    #[tokio::test]
467    async fn create_cache_uses_memory_for_unknown_backend() {
468        let cache = super::create_cache(&CacheConfig {
469            backend: "unknown".to_string(),
470            endpoint: "redis://127.0.0.1/".into(),
471            default_ttl: 5,
472        })
473        .await;
474
475        assert_eq!(cache.backend_name(), "memory");
476        cache.health_check().await.expect("memory cache is healthy");
477    }
478
479    #[cfg(feature = "memory")]
480    #[tokio::test]
481    async fn explicit_error_policy_rejects_unknown_backend() {
482        let Err(error) = super::create_cache_with_policy(
483            &CacheConfig {
484                backend: "unknown".to_string(),
485                ..CacheConfig::default()
486            },
487            CacheBackendFailurePolicy::ReturnError,
488        )
489        .await
490        else {
491            panic!("unknown backend should be returned to explicit callers");
492        };
493
494        assert!(
495            error
496                .to_string()
497                .contains("unsupported cache backend 'unknown'")
498        );
499    }
500
501    #[cfg(feature = "redis")]
502    #[tokio::test]
503    async fn explicit_error_policy_returns_redis_construction_error() {
504        let Err(error) = super::create_cache_with_policy(
505            &CacheConfig {
506                backend: "redis".to_string(),
507                endpoint: "not a redis url".into(),
508                default_ttl: 60,
509            },
510            CacheBackendFailurePolicy::ReturnError,
511        )
512        .await
513        else {
514            panic!("invalid Redis endpoint should be returned to explicit callers");
515        };
516
517        assert!(error.to_string().contains("redis cache connection"));
518    }
519
520    #[cfg(feature = "redis")]
521    #[test]
522    fn redis_backend_target_strips_credentials() {
523        assert_eq!(
524            super::redis_backend_target(&CacheEndpoint::url(
525                "redis://user:secret@example.com:6379/0",
526            )),
527            "redis://example.com:6379"
528        );
529    }
530
531    #[cfg(feature = "redis")]
532    #[test]
533    fn redis_backend_target_keeps_host_without_credentials() {
534        assert_eq!(
535            super::redis_backend_target(&CacheEndpoint::url("rediss://cache.internal:6380/1",)),
536            "rediss://cache.internal:6380"
537        );
538    }
539
540    #[cfg(feature = "redis")]
541    #[test]
542    fn redis_backend_target_handles_malformed_or_empty_hosts() {
543        assert_eq!(
544            super::redis_backend_target(&CacheEndpoint::url("not-a-url")),
545            "configured"
546        );
547        assert_eq!(
548            super::redis_backend_target(&CacheEndpoint::url("redis:///0")),
549            "redis://configured"
550        );
551    }
552
553    #[test]
554    fn cache_config_deserializes_structured_credentials_and_redacts_debug() {
555        let raw_username = "cache-user@example.com";
556        let raw_password = "cache#[]{}^+=*@:/?%secret";
557        let config: CacheConfig = serde_json::from_str(&format!(
558            r#"{{"backend":"redis","endpoint":{{"base_url":"redis://cache.example/0","username":"{raw_username}","password":"{raw_password}"}}}}"#,
559        ))
560        .unwrap();
561
562        assert_eq!(
563            config.endpoint,
564            CacheEndpoint::credentials(
565                "redis://cache.example/0",
566                Some(raw_username.to_string()),
567                Some(raw_password.to_string()),
568            )
569        );
570        let debug = format!("{config:?}");
571        assert!(!debug.contains(raw_username));
572        assert!(!debug.contains(raw_password));
573        assert!(debug.contains("CacheEndpoint::Credentials(<redacted>)"));
574
575        let serialized = serde_json::to_string(&config).unwrap();
576        assert!(serialized.contains("redis://cache.example/0"));
577        assert!(!serialized.contains(raw_username));
578        assert!(!serialized.contains(raw_password));
579        assert!(!serialized.contains("cache%23%5B%5D"));
580    }
581
582    #[test]
583    fn cache_error_display_messages_are_stable() {
584        assert_eq!(
585            CacheError::InvalidConfiguration("bad backend".to_string()).to_string(),
586            "invalid cache configuration: bad backend"
587        );
588        assert_eq!(
589            CacheError::RedisConnection("refused".to_string()).to_string(),
590            "redis cache connection: refused"
591        );
592        assert_eq!(
593            CacheError::RedisFallbackMode { remaining_ms: 25 }.to_string(),
594            "redis cache is in fallback mode for another 25ms"
595        );
596        assert_eq!(
597            CacheError::RedisHealthCheck("PONG missing".to_string()).to_string(),
598            "redis cache health check: PONG missing"
599        );
600        assert_eq!(
601            CacheError::RedisHealthCheckTimeout { timeout_ms: 250 }.to_string(),
602            "redis cache health check timed out after 250ms"
603        );
604    }
605}
aster_forge_cache/lib.rs

aster_forge_cache/
lib.rs
