distributed_cache/
runtime.rs

1//
2// Copyright 2018-2026 Accenture Technology
3//
4// Licensed under the Apache License, Version 2.0 (the "License");
5// you may not use this file except in compliance with the License.
6// You may obtain a copy of the License at
7//
8//     http://www.apache.org/licenses/LICENSE-2.0
9//
10// Unless required by applicable law or agreed to in writing, software
11// distributed under the License is distributed on an "AS IS" BASIS,
12// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13// See the License for the specific language governing permissions and
14// limitations under the License.
15//
16
17//! Process-wide holder of the cache's **single shared, multiplexed** backend
18//! and the [`RedisCacheStore`] over it — Rust port of the Java `CacheRuntime`.
19//! Every `v1.cache.redis` worker instance calls [`store`] and shares this one
20//! connection: `redis.cache.instances` is worker concurrency, not a connection
21//! count (design spec §4.4: no pool; the client pipelines over one in-order
22//! connection).
23//!
24//! The backend is built **lazily on first use**, re-resolving configuration
25//! on every attempt until it connects. This is deliberate (spec §6): the
26//! function is registered before a credential-bootstrap application publishes
27//! a vault password, and — unlike sync-over-async, which must keep an eager
28//! Pub/Sub subscriber live — a cache has nothing to maintain at start-up and
29//! **must not fail application start-up when Redis is briefly unreachable**.
30//! While the connection cannot be built the store stays empty and each call
31//! retries (a late credential is picked up); once built, the client owns
32//! reconnection under it and the store is reused.
33//!
34//! The one connection is released on shutdown through the platform's
35//! lifecycle ([`Platform::on_shutdown`]), registered from the build so the
36//! cleanup is wired only when a connection has actually been opened.
37
38use std::sync::{Arc, RwLock};
39
40use platform_core::{AppError, Platform};
41use redis_connection::RedisBackend;
42
43use crate::config::CacheConfig;
44use crate::store::RedisCacheStore;
45
46static STORE: RwLock<Option<Arc<RedisCacheStore>>> = RwLock::new(None);
47/// Serializes the slow build so concurrent first callers open ONE connection
48/// (the Java double-checked locking under a `ReentrantLock`).
49static BUILD: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
50
51/// The shared store, built on first use and reused thereafter. If the
52/// connection cannot be built yet, the error propagates to the caller
53/// (fail-fast — a cache failure is the caller's concern via the flow's
54/// exception handler) and the next call retries with freshly resolved
55/// configuration.
56pub async fn store() -> Result<Arc<RedisCacheStore>, AppError> {
57    if let Some(current) = current() {
58        return Ok(current);
59    }
60    let _guard = BUILD.lock().await;
61    if let Some(current) = current() {
62        return Ok(current);
63    }
64    let built = build().await?;
65    *STORE.write().unwrap_or_else(|p| p.into_inner()) = Some(built.clone());
66    Ok(built)
67}
68
69/// The store if one has been built (diagnostics; `None` before the first
70/// successful call and after [`shutdown`]).
71pub fn current() -> Option<Arc<RedisCacheStore>> {
72    STORE.read().unwrap_or_else(|p| p.into_inner()).clone()
73}
74
75/// Install a store as the process-wide instance — the reuse/test seam (a store
76/// built against an embedded or in-process server). Replaces any current one.
77pub fn set(store: Arc<RedisCacheStore>) {
78    *STORE.write().unwrap_or_else(|p| p.into_inner()) = Some(store);
79}
80
81/// Release the shared backend (idempotent) — registered with
82/// [`Platform::on_shutdown`] when the connection opens. Dropping the last
83/// handle closes the connection; a later call rebuilds from live configuration.
84pub fn shutdown() {
85    if STORE
86        .write()
87        .unwrap_or_else(|p| p.into_inner())
88        .take()
89        .is_some()
90    {
91        log::info!("Redis cache connection released");
92    }
93}
94
95async fn build() -> Result<Arc<RedisCacheStore>, AppError> {
96    let config = CacheConfig::from_config();
97    let redis = config.redis();
98    let backend = RedisBackend::connect(redis).await?;
99    // register cleanup now that a connection exists, via the platform's
100    // shutdown lifecycle. build() runs once under the lock per connection, so
101    // this registers exactly once per opened connection.
102    Platform::get_instance().on_shutdown(shutdown);
103    log::info!(
104        "Redis cache connected (redis {}, ssl={}, cluster={}, keyPrefix='{}', defaultTtl={}s)",
105        backend.endpoint(),
106        redis.ssl(),
107        backend.cluster(),
108        config.key_prefix(),
109        config.default_ttl_seconds()
110    );
111    Ok(Arc::new(RedisCacheStore::new(
112        backend,
113        config.key_prefix(),
114        config.default_ttl_seconds(),
115    )))
116}
distributed_cache/runtime.rs

distributed_cache/
runtime.rs
