rustlavel_cache/redis/
mod.rs

1//! A Redis client written from scratch on Tokio TCP.
2//!
3//! Three layers, each independently testable:
4//!
5//! * [`resp`] — the wire format. Pure functions over byte slices, so the
6//!   protocol tests need no server at all.
7//! * [`connection`] — one socket, the handshake, and command/reply framing.
8//! * [`pool`] — a bounded set of connections, discarding any that broke.
9//!
10//! On top sits [`RedisStore`], the [`Cache`] implementation. It uses only
11//! `GET`, `SET`, `DEL`, `EXISTS`, `INCRBY`, `DECRBY`, `FLUSHDB`, `EXPIRE`,
12//! `PEXPIRE`, `PTTL`, `PING` and `AUTH`/`SELECT` — a small enough surface that
13//! it also works against the Redis-compatible servers (Valkey, KeyDB,
14//! Dragonfly) people actually deploy.
15
16pub mod config;
17pub mod connection;
18pub mod pool;
19pub mod resp;
20
21pub use config::RedisConfig;
22pub use connection::Connection;
23pub use pool::{Pool, PooledConnection};
24pub use resp::Value;
25
26use crate::store::{BoxFuture, Cache, decode, prefixed, record};
27use rustlavel_core::{Error, Json, Result};
28use std::time::Duration;
29
30/// A cache backed by Redis.
31///
32/// Cloning shares one pool, so this can be registered as application state.
33#[derive(Clone)]
34pub struct RedisStore {
35    pool: Pool,
36    prefix: String,
37}
38
39impl RedisStore {
40    /// Build a store from a URL: `redis://[:password@]host:port[/db]`.
41    pub fn connect(url: &str) -> Result<Self> {
42        Ok(RedisStore::new(RedisConfig::from_url(url)?, ""))
43    }
44
45    pub fn new(config: RedisConfig, prefix: impl Into<String>) -> Self {
46        RedisStore { pool: Pool::new(config), prefix: prefix.into() }
47    }
48
49    pub fn pool(&self) -> &Pool {
50        &self.pool
51    }
52
53    /// Open a connection now, so a bad URL or a wrong password surfaces at boot.
54    pub async fn verify(&self) -> Result<()> {
55        self.pool.verify().await
56    }
57
58    /// `PING`. Returns `PONG`, and is the cheapest liveness check there is.
59    pub async fn ping(&self) -> Result<String> {
60        let reply = self.pool.command(&[b"PING"]).await?.into_result()?;
61        reply
62            .as_str()
63            .map(str::to_string)
64            .ok_or_else(|| Error::msg("Redis answered PING with something other than a status"))
65    }
66
67    /// `EXPIRE key seconds` — whole-second precision, which is what the Redis
68    /// command itself offers. Returns whether the key existed.
69    pub async fn expire(&self, key: &str, seconds: u64) -> Result<bool> {
70        let full = prefixed(&self.prefix, key);
71        let seconds = seconds.to_string();
72        let reply = self
73            .pool
74            .command(&[b"EXPIRE", full.as_bytes(), seconds.as_bytes()])
75            .await?
76            .into_result()?;
77        Ok(reply.as_i64() == Some(1))
78    }
79
80    /// `PEXPIRE key milliseconds`, for the sub-second windows a rate limiter
81    /// wants and `EXPIRE` cannot express.
82    pub async fn pexpire(&self, key: &str, ttl: Duration) -> Result<bool> {
83        let full = prefixed(&self.prefix, key);
84        let millis = (ttl.as_millis() as u64).max(1).to_string();
85        let reply = self
86            .pool
87            .command(&[b"PEXPIRE", full.as_bytes(), millis.as_bytes()])
88            .await?
89            .into_result()?;
90        Ok(reply.as_i64() == Some(1))
91    }
92
93    /// Run an arbitrary command. The escape hatch for anything this crate does
94    /// not wrap; arguments are still length-prefixed, so it cannot be injected.
95    pub async fn command(&self, args: &[&[u8]]) -> Result<Value> {
96        self.pool.command(args).await?.into_result()
97    }
98
99    async fn integer(&self, args: &[&[u8]]) -> Result<i64> {
100        let reply = self.pool.command(args).await?.into_result()?;
101        reply.as_i64().ok_or_else(|| {
102            Error::msg(format!("expected an integer reply from Redis, got {reply:?}"))
103        })
104    }
105}
106
107impl Cache for RedisStore {
108    fn driver(&self) -> &'static str {
109        "redis"
110    }
111
112    fn get<'a>(&'a self, key: &'a str) -> BoxFuture<'a, Result<Option<Json>>> {
113        Box::pin(async move {
114            let full = prefixed(&self.prefix, key);
115            let reply = self.pool.command(&[b"GET", full.as_bytes()]).await?.into_result()?;
116
117            // Redis expires keys for us, so a nil reply is the whole miss path.
118            let found = reply.as_str().and_then(decode);
119            record(found.is_some(), "redis", key);
120            Ok(found)
121        })
122    }
123
124    fn put<'a>(&'a self, key: &'a str, value: Json, ttl: Duration) -> BoxFuture<'a, Result<()>> {
125        Box::pin(async move {
126            let full = prefixed(&self.prefix, key);
127            if ttl.is_zero() {
128                self.pool.command(&[b"DEL", full.as_bytes()]).await?.into_result()?;
129                return Ok(());
130            }
131
132            // PX rather than EX: a caller asking for 500ms should get 500ms,
133            // not a second rounded either way.
134            let millis = (ttl.as_millis() as u64).max(1).to_string();
135            let payload = value.to_string();
136            self.pool
137                .command(&[b"SET", full.as_bytes(), payload.as_bytes(), b"PX", millis.as_bytes()])
138                .await?
139                .into_result()?;
140            Ok(())
141        })
142    }
143
144    fn forever<'a>(&'a self, key: &'a str, value: Json) -> BoxFuture<'a, Result<()>> {
145        Box::pin(async move {
146            let full = prefixed(&self.prefix, key);
147            let payload = value.to_string();
148            self.pool
149                .command(&[b"SET", full.as_bytes(), payload.as_bytes()])
150                .await?
151                .into_result()?;
152            Ok(())
153        })
154    }
155
156    fn forget<'a>(&'a self, key: &'a str) -> BoxFuture<'a, Result<bool>> {
157        Box::pin(async move {
158            let full = prefixed(&self.prefix, key);
159            Ok(self.integer(&[b"DEL", full.as_bytes()]).await? > 0)
160        })
161    }
162
163    fn flush(&self) -> BoxFuture<'_, Result<()>> {
164        Box::pin(async move {
165            // `FLUSHDB` empties the whole database, prefix or no prefix — there
166            // is no server-side "delete by prefix" that is safe on a large
167            // keyspace (`KEYS` blocks the server). Give a cache that shares a
168            // Redis with anything else its own database number.
169            self.pool.command(&[b"FLUSHDB"]).await?.into_result()?;
170            Ok(())
171        })
172    }
173
174    fn has<'a>(&'a self, key: &'a str) -> BoxFuture<'a, Result<bool>> {
175        Box::pin(async move {
176            let full = prefixed(&self.prefix, key);
177            Ok(self.integer(&[b"EXISTS", full.as_bytes()]).await? > 0)
178        })
179    }
180
181    fn increment<'a>(&'a self, key: &'a str, by: i64) -> BoxFuture<'a, Result<i64>> {
182        Box::pin(async move {
183            let full = prefixed(&self.prefix, key);
184            let by = by.to_string();
185            self.integer(&[b"INCRBY", full.as_bytes(), by.as_bytes()]).await
186        })
187    }
188
189    fn decrement<'a>(&'a self, key: &'a str, by: i64) -> BoxFuture<'a, Result<i64>> {
190        Box::pin(async move {
191            let full = prefixed(&self.prefix, key);
192            let by = by.to_string();
193            self.integer(&[b"DECRBY", full.as_bytes(), by.as_bytes()]).await
194        })
195    }
196
197    fn increment_within<'a>(
198        &'a self,
199        key: &'a str,
200        by: i64,
201        ttl: Duration,
202    ) -> BoxFuture<'a, Result<i64>> {
203        Box::pin(async move {
204            let full = prefixed(&self.prefix, key);
205            let millis = (ttl.as_millis() as u64).max(1).to_string();
206
207            // `SET key 0 PX ttl NX` creates the counter *with* its window and
208            // does nothing at all if it already exists, so a later request in
209            // the same window cannot extend it. Then `INCRBY` counts. Two
210            // round trips, no lost TTL, and no read-modify-write race: the
211            // `NX` and the `INCRBY` are each atomic on the server.
212            self.pool
213                .command(&[b"SET", full.as_bytes(), b"0", b"PX", millis.as_bytes(), b"NX"])
214                .await?
215                .into_result()?;
216
217            let by = by.to_string();
218            self.integer(&[b"INCRBY", full.as_bytes(), by.as_bytes()]).await
219        })
220    }
221
222    fn ttl<'a>(&'a self, key: &'a str) -> BoxFuture<'a, Result<Option<Duration>>> {
223        Box::pin(async move {
224            let full = prefixed(&self.prefix, key);
225            // PTTL answers -2 for a missing key and -1 for one with no expiry;
226            // both mean "no deadline to report".
227            let millis = self.integer(&[b"PTTL", full.as_bytes()]).await?;
228            Ok((millis >= 0).then(|| Duration::from_millis(millis as u64)))
229        })
230    }
231}
rustlavel_cache/redis/mod.rs

rustlavel_cache/redis/
mod.rs
