nidus_redis/
lib.rs

1#![deny(missing_docs)]
2
3//! First-party Redis adapter for Nidus applications.
4//!
5//! The provider exposes both [`redis::Client`] and the reconnecting
6//! [`redis::aio::ConnectionManager`]. Nidus-owned convenience operations are
7//! bounded and instrumented; raw Redis commands remain fully available and
8//! retain native Redis semantics.
9
10use std::{
11    fmt,
12    sync::{
13        Arc,
14        atomic::{AtomicBool, Ordering},
15    },
16    time::{Duration, Instant},
17};
18
19use async_trait::async_trait;
20use nidus_core::{Container, LifecycleHook, NidusError};
21use nidus_integrations::{IntegrationEvent, IntegrationStatus, IntegrationTelemetry};
22use thiserror::Error;
23use tokio::sync::{Semaphore, SemaphorePermit};
24
25const DEFAULT_CONCURRENCY_LIMIT: usize = 256;
26const MAX_CONCURRENCY_LIMIT: usize = 65_536;
27
28/// Result type used by Redis adapter operations.
29pub type Result<T> = std::result::Result<T, RedisError>;
30
31/// Error returned by Redis adapter operations.
32#[derive(Debug, Error)]
33pub enum RedisError {
34    /// The Redis client or server returned an error.
35    #[error(transparent)]
36    Redis(#[from] redis::RedisError),
37    /// Nidus provider registration failed.
38    #[error(transparent)]
39    Nidus(#[from] NidusError),
40    /// Nidus config deserialization failed.
41    #[cfg(feature = "nidus-config")]
42    #[error(transparent)]
43    Config(#[from] nidus_config::ConfigError),
44    /// A required bound was configured as zero.
45    #[error("{field} must be greater than zero")]
46    InvalidBound {
47        /// Invalid configuration field.
48        field: &'static str,
49    },
50    /// The connection URL is not safe for the selected environment.
51    #[error("invalid Redis configuration: {message}")]
52    Configuration {
53        /// Redaction-safe validation message.
54        message: &'static str,
55    },
56    /// The provider stopped accepting adapter-owned work during shutdown.
57    #[error("Redis provider is shutting down")]
58    ShuttingDown,
59    /// Admitted Redis operations did not drain before the shutdown deadline.
60    #[error("Redis operations did not drain before the shutdown timeout")]
61    ShutdownTimeout,
62}
63
64/// Typed Redis connection and reconnect configuration.
65#[derive(Clone, Eq, PartialEq)]
66pub struct RedisConfig {
67    url: String,
68    connection_timeout: Duration,
69    response_timeout: Duration,
70    shutdown_timeout: Duration,
71    reconnect_attempts: usize,
72    reconnect_min_delay: Duration,
73    reconnect_max_delay: Duration,
74    concurrency_limit: usize,
75    pipeline_buffer_size: usize,
76    allow_plaintext_local: bool,
77}
78
79impl RedisConfig {
80    /// Creates secure, bounded Redis configuration from an explicit URL.
81    pub fn new(url: impl Into<String>) -> Self {
82        Self {
83            url: url.into(),
84            connection_timeout: Duration::from_secs(5),
85            response_timeout: Duration::from_secs(2),
86            shutdown_timeout: Duration::from_secs(5),
87            reconnect_attempts: 6,
88            reconnect_min_delay: Duration::from_millis(100),
89            reconnect_max_delay: Duration::from_secs(5),
90            concurrency_limit: DEFAULT_CONCURRENCY_LIMIT,
91            pipeline_buffer_size: 1024,
92            allow_plaintext_local: false,
93        }
94    }
95
96    /// Sets the timeout for initial and reconnect attempts.
97    pub fn with_connection_timeout(mut self, timeout: Duration) -> Self {
98        self.connection_timeout = timeout;
99        self
100    }
101
102    /// Sets the timeout for command responses.
103    pub fn with_response_timeout(mut self, timeout: Duration) -> Self {
104        self.response_timeout = timeout;
105        self
106    }
107
108    /// Sets how long graceful shutdown waits for admitted operations.
109    pub fn with_shutdown_timeout(mut self, timeout: Duration) -> Self {
110        self.shutdown_timeout = timeout;
111        self
112    }
113
114    /// Sets the maximum reconnect attempts performed by redis-rs.
115    pub fn with_reconnect_attempts(mut self, attempts: usize) -> Self {
116        self.reconnect_attempts = attempts;
117        self
118    }
119
120    /// Sets minimum and maximum exponential reconnect delays.
121    pub fn with_reconnect_backoff(mut self, minimum: Duration, maximum: Duration) -> Self {
122        self.reconnect_min_delay = minimum;
123        self.reconnect_max_delay = maximum;
124        self
125    }
126
127    /// Sets the maximum concurrent requests accepted by the connection manager.
128    pub fn with_concurrency_limit(mut self, limit: usize) -> Self {
129        self.concurrency_limit = limit;
130        self
131    }
132
133    /// Sets the bounded internal pipeline buffer size.
134    pub fn with_pipeline_buffer_size(mut self, size: usize) -> Self {
135        self.pipeline_buffer_size = size;
136        self
137    }
138
139    /// Explicitly permits `redis://` only for a loopback development server.
140    pub fn allow_plaintext_for_local_development(mut self) -> Self {
141        self.allow_plaintext_local = true;
142        self
143    }
144
145    /// Returns the configured Redis URL.
146    pub fn url(&self) -> &str {
147        &self.url
148    }
149
150    /// Validates all non-zero bounds before connecting.
151    pub fn validate(&self) -> Result<()> {
152        for (field, value) in [
153            ("connection_timeout", self.connection_timeout.as_nanos()),
154            ("response_timeout", self.response_timeout.as_nanos()),
155            ("shutdown_timeout", self.shutdown_timeout.as_nanos()),
156            ("reconnect_min_delay", self.reconnect_min_delay.as_nanos()),
157            ("reconnect_max_delay", self.reconnect_max_delay.as_nanos()),
158            ("concurrency_limit", self.concurrency_limit as u128),
159            ("pipeline_buffer_size", self.pipeline_buffer_size as u128),
160        ] {
161            if value == 0 {
162                return Err(RedisError::InvalidBound { field });
163            }
164        }
165        if self.concurrency_limit > MAX_CONCURRENCY_LIMIT {
166            return Err(RedisError::Configuration {
167                message: "Redis concurrency_limit must not exceed 65536",
168            });
169        }
170        if self.reconnect_max_delay < self.reconnect_min_delay {
171            return Err(RedisError::Configuration {
172                message: "Redis reconnect delays must be ordered",
173            });
174        }
175        if self.url.starts_with("redis://") {
176            if !self.allow_plaintext_local || !redis_url_is_loopback(&self.url) {
177                return Err(RedisError::Configuration {
178                    message: "plaintext Redis is restricted to explicit loopback development URLs",
179                });
180            }
181        } else if !self.url.starts_with("rediss://") && !self.url.starts_with("redis+unix://") {
182            return Err(RedisError::Configuration {
183                message: "production Redis URLs must use rediss:// or a Unix socket",
184            });
185        }
186        Ok(())
187    }
188
189    /// Loads Redis config from a nested `nidus_config::Config` path.
190    #[cfg(feature = "nidus-config")]
191    pub fn from_config_path<I, S>(config: &nidus_config::Config, path: I) -> Result<Self>
192    where
193        I: IntoIterator<Item = S>,
194        S: AsRef<str>,
195    {
196        #[derive(serde::Deserialize)]
197        struct RawConfig {
198            url: String,
199            connection_timeout_ms: Option<u64>,
200            response_timeout_ms: Option<u64>,
201            shutdown_timeout_ms: Option<u64>,
202            reconnect_attempts: Option<usize>,
203            concurrency_limit: Option<usize>,
204            pipeline_buffer_size: Option<usize>,
205            allow_plaintext_local: Option<bool>,
206        }
207
208        let raw: RawConfig = config.get_required_path_typed(path)?;
209        let mut settings = Self::new(raw.url);
210        if let Some(value) = raw.connection_timeout_ms {
211            settings = settings.with_connection_timeout(Duration::from_millis(value));
212        }
213        if let Some(value) = raw.response_timeout_ms {
214            settings = settings.with_response_timeout(Duration::from_millis(value));
215        }
216        if let Some(value) = raw.shutdown_timeout_ms {
217            settings = settings.with_shutdown_timeout(Duration::from_millis(value));
218        }
219        if let Some(value) = raw.reconnect_attempts {
220            settings = settings.with_reconnect_attempts(value);
221        }
222        if let Some(value) = raw.concurrency_limit {
223            settings = settings.with_concurrency_limit(value);
224        }
225        if let Some(value) = raw.pipeline_buffer_size {
226            settings = settings.with_pipeline_buffer_size(value);
227        }
228        if raw.allow_plaintext_local == Some(true) {
229            settings = settings.allow_plaintext_for_local_development();
230        }
231        settings.validate()?;
232        Ok(settings)
233    }
234}
235
236impl fmt::Debug for RedisConfig {
237    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
238        formatter
239            .debug_struct("RedisConfig")
240            .field("url", &"<redacted>")
241            .field("connection_timeout", &self.connection_timeout)
242            .field("response_timeout", &self.response_timeout)
243            .field("shutdown_timeout", &self.shutdown_timeout)
244            .field("reconnect_attempts", &self.reconnect_attempts)
245            .field("reconnect_min_delay", &self.reconnect_min_delay)
246            .field("reconnect_max_delay", &self.reconnect_max_delay)
247            .field("concurrency_limit", &self.concurrency_limit)
248            .field("pipeline_buffer_size", &self.pipeline_buffer_size)
249            .field("allow_plaintext_local", &self.allow_plaintext_local)
250            .finish()
251    }
252}
253
254fn redis_url_is_loopback(url: &str) -> bool {
255    url::Url::parse(url).ok().is_some_and(|url| {
256        url.scheme() == "redis"
257            && match url.host() {
258                Some(url::Host::Domain(host)) => {
259                    host.eq_ignore_ascii_case("localhost")
260                        || host
261                            .parse::<std::net::IpAddr>()
262                            .is_ok_and(|address| address.is_loopback())
263                }
264                Some(url::Host::Ipv4(host)) => host.is_loopback(),
265                Some(url::Host::Ipv6(host)) => host.is_loopback(),
266                None => false,
267            }
268    })
269}
270
271/// Builder for a reconnecting Redis provider.
272#[derive(Clone, Debug)]
273pub struct RedisProviderBuilder {
274    config: RedisConfig,
275    telemetry: IntegrationTelemetry,
276}
277
278impl RedisProviderBuilder {
279    /// Creates a builder from an explicit Redis URL.
280    pub fn new(url: impl Into<String>) -> Self {
281        Self {
282            config: RedisConfig::new(url),
283            telemetry: IntegrationTelemetry::new(),
284        }
285    }
286
287    /// Replaces the typed Redis configuration.
288    pub fn config(mut self, config: RedisConfig) -> Self {
289        self.config = config;
290        self
291    }
292
293    /// Adds shared tracing, metrics, dashboard, or custom telemetry.
294    pub fn telemetry(mut self, telemetry: IntegrationTelemetry) -> Self {
295        self.telemetry = telemetry;
296        self
297    }
298
299    /// Builds the raw Redis client without performing network I/O.
300    pub fn build_client(&self) -> Result<redis::Client> {
301        self.config.validate()?;
302        Ok(redis::Client::open(self.config.url.as_str())?)
303    }
304
305    /// Connects and returns a provider wrapping native Redis clients.
306    pub async fn connect(self) -> Result<RedisProvider> {
307        self.config.validate()?;
308        let started_at = Instant::now();
309        let client = redis::Client::open(self.config.url.as_str())?;
310        let manager_config = redis::aio::ConnectionManagerConfig::new()
311            .set_number_of_retries(self.config.reconnect_attempts)
312            .set_min_delay(self.config.reconnect_min_delay)
313            .set_max_delay(self.config.reconnect_max_delay)
314            .set_response_timeout(Some(self.config.response_timeout))
315            .set_connection_timeout(Some(self.config.connection_timeout))
316            .set_concurrency_limit(self.config.concurrency_limit)
317            .set_pipeline_buffer_size(self.config.pipeline_buffer_size);
318        let result =
319            redis::aio::ConnectionManager::new_with_config(client.clone(), manager_config).await;
320        self.telemetry
321            .record(&IntegrationEvent::new(
322                "nidus-redis",
323                "connect",
324                if result.is_ok() {
325                    IntegrationStatus::Success
326                } else {
327                    IntegrationStatus::Failure
328                },
329                started_at.elapsed(),
330            ))
331            .await;
332        Ok(RedisProvider {
333            client,
334            connection: result?,
335            in_flight: Arc::new(Semaphore::new(self.config.concurrency_limit)),
336            max_in_flight: self.config.concurrency_limit as u32,
337            shutdown_timeout: self.config.shutdown_timeout,
338            shutting_down: Arc::new(AtomicBool::new(false)),
339            shutdown_complete: Arc::new(AtomicBool::new(false)),
340            telemetry: self.telemetry,
341        })
342    }
343
344    /// Connects and registers the provider as a Nidus singleton.
345    pub async fn register(self, container: &mut Container) -> Result<()> {
346        container.register_singleton(self.connect().await?)?;
347        Ok(())
348    }
349}
350
351/// Nidus provider exposing native Redis clients and bounded conveniences.
352#[derive(Clone)]
353pub struct RedisProvider {
354    client: redis::Client,
355    connection: redis::aio::ConnectionManager,
356    in_flight: Arc<Semaphore>,
357    max_in_flight: u32,
358    shutdown_timeout: Duration,
359    shutting_down: Arc<AtomicBool>,
360    shutdown_complete: Arc<AtomicBool>,
361    telemetry: IntegrationTelemetry,
362}
363
364impl fmt::Debug for RedisProvider {
365    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
366        formatter
367            .debug_struct("RedisProvider")
368            .field("client", &"redis::Client(<redacted>)")
369            .field("connection", &"redis::aio::ConnectionManager")
370            .field("shutting_down", &self.shutting_down.load(Ordering::Acquire))
371            .field("available_permits", &self.in_flight.available_permits())
372            .field("telemetry", &self.telemetry)
373            .finish()
374    }
375}
376
377impl RedisProvider {
378    /// Creates a Redis provider builder.
379    pub fn builder(url: impl Into<String>) -> RedisProviderBuilder {
380        RedisProviderBuilder::new(url)
381    }
382
383    /// Creates a provider from existing native Redis clients.
384    pub fn from_parts(client: redis::Client, connection: redis::aio::ConnectionManager) -> Self {
385        Self {
386            client,
387            connection,
388            in_flight: Arc::new(Semaphore::new(DEFAULT_CONCURRENCY_LIMIT)),
389            max_in_flight: DEFAULT_CONCURRENCY_LIMIT as u32,
390            shutdown_timeout: Duration::from_secs(5),
391            shutting_down: Arc::new(AtomicBool::new(false)),
392            shutdown_complete: Arc::new(AtomicBool::new(false)),
393            telemetry: IntegrationTelemetry::new(),
394        }
395    }
396
397    /// Returns the native Redis client.
398    pub fn client(&self) -> &redis::Client {
399        &self.client
400    }
401
402    /// Returns the reconnecting, cloneable native connection manager.
403    pub fn connection_manager(&self) -> &redis::aio::ConnectionManager {
404        &self.connection
405    }
406
407    /// Returns a clone of the reconnecting native connection manager.
408    pub fn connection(&self) -> redis::aio::ConnectionManager {
409        self.connection.clone()
410    }
411
412    /// Gets a binary value using a Nidus-observed Redis operation.
413    pub async fn get(&self, key: &str) -> Result<Option<Vec<u8>>> {
414        let _permit = self.acquire_permit().await?;
415        let started_at = Instant::now();
416        let mut connection = self.connection.clone();
417        let result = redis::cmd("GET")
418            .arg(key)
419            .query_async(&mut connection)
420            .await;
421        self.record("get", result.is_ok(), started_at).await;
422        Ok(result?)
423    }
424
425    /// Sets a binary value, optionally with a positive TTL.
426    pub async fn set(&self, key: &str, value: &[u8], ttl: Option<Duration>) -> Result<()> {
427        if let Some(ttl) = ttl
428            && ttl.is_zero()
429        {
430            return Err(RedisError::InvalidBound { field: "ttl" });
431        }
432        let _permit = self.acquire_permit().await?;
433        let started_at = Instant::now();
434        let mut command = redis::cmd("SET");
435        command.arg(key).arg(value);
436        if let Some(ttl) = ttl {
437            command
438                .arg("PX")
439                .arg(ttl.as_millis().min(u128::from(u64::MAX)) as u64);
440        }
441        let mut connection = self.connection.clone();
442        let result: redis::RedisResult<()> = command.query_async(&mut connection).await;
443        self.record("set", result.is_ok(), started_at).await;
444        Ok(result?)
445    }
446
447    /// Deletes one key and returns whether it existed.
448    pub async fn delete(&self, key: &str) -> Result<bool> {
449        let _permit = self.acquire_permit().await?;
450        let started_at = Instant::now();
451        let mut connection = self.connection.clone();
452        let result: redis::RedisResult<u64> = redis::cmd("DEL")
453            .arg(key)
454            .query_async(&mut connection)
455            .await;
456        self.record("delete", result.is_ok(), started_at).await;
457        Ok(result? > 0)
458    }
459
460    /// Executes a lightweight Redis `PING` readiness check.
461    #[cfg(feature = "health")]
462    pub async fn health_status(&self) -> nidus_http::health::HealthStatus {
463        let Ok(_permit) = self.acquire_permit().await else {
464            return nidus_http::health::HealthStatus::down("redis provider is shutting down");
465        };
466        let started_at = Instant::now();
467        let mut connection = self.connection.clone();
468        let result: redis::RedisResult<String> =
469            redis::cmd("PING").query_async(&mut connection).await;
470        self.record("health", result.as_deref() == Ok("PONG"), started_at)
471            .await;
472        if result.as_deref() == Ok("PONG") {
473            nidus_http::health::HealthStatus::up()
474        } else {
475            nidus_http::health::HealthStatus::down("redis readiness check failed")
476        }
477    }
478
479    /// Adds this provider as a readiness check on a health registry.
480    #[cfg(feature = "health")]
481    pub fn register_ready_check(
482        self: std::sync::Arc<Self>,
483        registry: nidus_http::health::HealthRegistry,
484        name: impl Into<String>,
485    ) -> nidus_http::health::HealthRegistry {
486        registry.ready_check(name, move || {
487            let provider = std::sync::Arc::clone(&self);
488            async move { provider.health_status().await }
489        })
490    }
491
492    /// Stops new adapter-owned work and waits for admitted operations to drain.
493    ///
494    /// Existing native client clones remain under application ownership.
495    pub async fn shutdown(&self) -> Result<()> {
496        if self.shutdown_complete.load(Ordering::Acquire) {
497            return Ok(());
498        }
499        self.shutting_down.store(true, Ordering::Release);
500        let drained = tokio::time::timeout(
501            self.shutdown_timeout,
502            self.in_flight.acquire_many(self.max_in_flight),
503        )
504        .await
505        .map_err(|_| RedisError::ShutdownTimeout)?
506        .map_err(|_| RedisError::ShuttingDown)?;
507        self.shutdown_complete.store(true, Ordering::Release);
508        self.in_flight.close();
509        drop(drained);
510        Ok(())
511    }
512
513    async fn acquire_permit(&self) -> Result<SemaphorePermit<'_>> {
514        if self.shutting_down.load(Ordering::Acquire) {
515            return Err(RedisError::ShuttingDown);
516        }
517        let permit = self
518            .in_flight
519            .acquire()
520            .await
521            .map_err(|_| RedisError::ShuttingDown)?;
522        if self.shutting_down.load(Ordering::Acquire) {
523            return Err(RedisError::ShuttingDown);
524        }
525        Ok(permit)
526    }
527
528    async fn record(&self, operation: &'static str, success: bool, started_at: Instant) {
529        self.telemetry
530            .record(&IntegrationEvent::new(
531                "nidus-redis",
532                operation,
533                if success {
534                    IntegrationStatus::Success
535                } else {
536                    IntegrationStatus::Failure
537                },
538                started_at.elapsed(),
539            ))
540            .await;
541    }
542}
543
544#[async_trait]
545impl LifecycleHook for RedisProvider {
546    async fn on_shutdown(&self) -> nidus_core::Result<()> {
547        self.shutdown()
548            .await
549            .map_err(|_| NidusError::ApplicationBuild {
550                message: "Redis operations failed to drain during shutdown".to_owned(),
551            })
552    }
553}
nidus_redis/lib.rs

nidus_redis/
lib.rs
