featherbit/stores/
redis_store.rs

1//! Redis/Valkey client for named stores (`redis-store` feature).
2//!
3//! One [`RedisStoreClient`] per `stores:` entry: env placeholders in `url` /
4//! `password` resolve here (point of use — the stored config stays raw), the
5//! underlying connection is a single auto-reconnecting multiplexed
6//! `ConnectionManager` created lazily on first use and handed out as a
7//! [`StoreConn`] that bounds every command by `connect_budget`, and a
8//! resolved-config fingerprint lets [`super::StoreRegistry::rebuild`] keep
9//! the connection across unrelated config reloads.
10
11use std::time::Duration;
12
13use base64::engine::general_purpose::STANDARD as BASE64;
14use base64::Engine;
15use redis::IntoConnectionInfo;
16use ring::digest::{digest, SHA256};
17use tokio::sync::OnceCell;
18
19use crate::config::interpolate_env;
20use crate::config::StoreConfig;
21
22/// The connection handed to every store consumer.
23///
24/// A thin wrapper over the shared `ConnectionManager` whose every command is
25/// bounded by the store's `connect_budget`. The manager alone bounds only the
26/// *first* connection: once the store goes away mid-life, each command awaits
27/// the manager's shared reconnect future -- six exponentially backed-off
28/// attempts, each up to `connect_timeout` -- and nothing in the crate caps
29/// that total. Measured against a stopped container, the second request
30/// after an outage held its worker for ~55s before the `503` the design
31/// promises. The budget is the one figure an operator can reason about, so it
32/// bounds every wait for the store, not just the opening one.
33#[derive(Clone)]
34pub struct StoreConn {
35    inner: redis::aio::ConnectionManager,
36    budget: Duration,
37    name: String,
38}
39
40impl StoreConn {
41    fn gave_up(&self) -> redis::RedisError {
42        redis::RedisError::from((
43            redis::ErrorKind::IoError,
44            "store operation gave up",
45            format!(
46                "store '{}': no connection within {}ms (connect_budget_ms)",
47                self.name,
48                self.budget.as_millis()
49            ),
50        ))
51    }
52}
53
54impl redis::aio::ConnectionLike for StoreConn {
55    fn req_packed_command<'a>(
56        &'a mut self,
57        cmd: &'a redis::Cmd,
58    ) -> redis::RedisFuture<'a, redis::Value> {
59        Box::pin(async move {
60            match tokio::time::timeout(self.budget, self.inner.req_packed_command(cmd)).await {
61                Ok(result) => result,
62                Err(_) => Err(self.gave_up()),
63            }
64        })
65    }
66
67    fn req_packed_commands<'a>(
68        &'a mut self,
69        cmd: &'a redis::Pipeline,
70        offset: usize,
71        count: usize,
72    ) -> redis::RedisFuture<'a, Vec<redis::Value>> {
73        Box::pin(async move {
74            match tokio::time::timeout(
75                self.budget,
76                self.inner.req_packed_commands(cmd, offset, count),
77            )
78            .await
79            {
80                Ok(result) => result,
81                Err(_) => Err(self.gave_up()),
82            }
83        })
84    }
85
86    fn get_db(&self) -> i64 {
87        self.inner.get_db()
88    }
89}
90
91/// Result of a connectivity check (`POST /api/stores/{name}/ping`).
92pub struct PingInfo {
93    pub latency_ms: u64,
94    /// `valkey_version` when the server is Valkey, else `redis_version`.
95    pub version: String,
96}
97
98pub struct RedisStoreClient {
99    name: String,
100    key_prefix: String,
101    fingerprint: String,
102    connect_timeout: Duration,
103    connect_budget: Duration,
104    client: redis::Client,
105    conn: OnceCell<redis::aio::ConnectionManager>,
106}
107
108// `redis::aio::ConnectionManager` has no `Debug` impl, so this can't be
109// `#[derive(Debug)]`'d; a manual impl covering the non-connection fields is
110// enough for test assertions (`Result::unwrap_err` requires `T: Debug`) and
111// avoids ever printing connection internals or credentials.
112impl std::fmt::Debug for RedisStoreClient {
113    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
114        f.debug_struct("RedisStoreClient")
115            .field("name", &self.name)
116            .field("key_prefix", &self.key_prefix)
117            .field("connect_timeout", &self.connect_timeout)
118            .finish_non_exhaustive()
119    }
120}
121
122impl RedisStoreClient {
123    /// Builds the client: resolves `${ENV}` in url/password, parses the URL,
124    /// loads the CA bundle if configured. **No network I/O** — connection is
125    /// deferred to [`Self::conn`], so config apply never blocks on a store.
126    pub fn build(cfg: &StoreConfig) -> Result<Self, String> {
127        let url = interpolate_env(&cfg.url);
128        if url.trim().is_empty() {
129            return Err(format!(
130                "store '{}': url resolved to an empty string",
131                cfg.name
132            ));
133        }
134        let mut info = url
135            .as_str()
136            .into_connection_info()
137            .map_err(|e| format!("store '{}': invalid url: {}", cfg.name, e))?;
138        if let Some(pw) = cfg.password.as_deref() {
139            let pw = interpolate_env(pw);
140            if !pw.is_empty() {
141                info.redis.password = Some(pw);
142            }
143        }
144        let client = match cfg.tls.as_ref().and_then(|t| t.ca_cert_path.as_deref()) {
145            Some(path) => {
146                let pem = std::fs::read(path).map_err(|e| {
147                    format!(
148                        "store '{}': cannot read ca_cert_path '{}': {}",
149                        cfg.name, path, e
150                    )
151                })?;
152                redis::Client::build_with_tls(
153                    info,
154                    redis::TlsCertificates {
155                        client_tls: None,
156                        root_cert: Some(pem),
157                    },
158                )
159                .map_err(|e| format!("store '{}': tls setup: {}", cfg.name, e))?
160            }
161            None => {
162                redis::Client::open(info).map_err(|e| format!("store '{}': {}", cfg.name, e))?
163            }
164        };
165        Ok(Self {
166            name: cfg.name.clone(),
167            key_prefix: cfg.key_prefix.clone(),
168            fingerprint: Self::fingerprint_of(cfg),
169            connect_timeout: Duration::from_millis(cfg.connect_timeout_ms),
170            connect_budget: Duration::from_millis(cfg.connect_budget_ms),
171            client,
172            conn: OnceCell::new(),
173        })
174    }
175
176    /// Fingerprint over the *resolved* connection-relevant fields, so a
177    /// changed env var (not just changed YAML) rebuilds the client on the
178    /// next reload. Hashed so no secret sits in an easily-dumped string.
179    pub fn fingerprint_of(cfg: &StoreConfig) -> String {
180        let material = format!(
181            "{}|{}|{}|{}|{}",
182            interpolate_env(&cfg.url),
183            cfg.password
184                .as_deref()
185                .map(interpolate_env)
186                .unwrap_or_default(),
187            cfg.key_prefix,
188            cfg.connect_timeout_ms,
189            cfg.tls
190                .as_ref()
191                .and_then(|t| t.ca_cert_path.as_deref())
192                .unwrap_or(""),
193        );
194        BASE64.encode(digest(&SHA256, material.as_bytes()))
195    }
196
197    /// The shared multiplexed connection; established on first use and
198    /// auto-reconnecting thereafter. Every command on it is bounded by
199    /// `connect_budget` -- see [`StoreConn`].
200    pub async fn conn(&self) -> Result<StoreConn, String> {
201        let manager = self
202            .conn
203            .get_or_try_init(|| async {
204                let cfg = redis::aio::ConnectionManagerConfig::new()
205                    .set_connection_timeout(self.connect_timeout)
206                    .set_response_timeout(self.connect_timeout)
207                    // No individual backoff may outlast the budget itself.
208                    .set_max_delay(self.connect_budget.as_millis() as u64);
209
210                // `connect_timeout` bounds one attempt; it says nothing about
211                // the retry schedule around them. With the crate defaults that
212                // is six attempts separated by 100/200/400/800/1600/3200ms of
213                // backoff, none of it capped -- so an unreachable store held
214                // the request open for tens of seconds before the `503` the
215                // design promises, on the first request after an outage began.
216                // The budget bounds connect + retries + waits together, which
217                // is the only figure an operator can actually reason about.
218                match tokio::time::timeout(
219                    self.connect_budget,
220                    redis::aio::ConnectionManager::new_with_config(self.client.clone(), cfg),
221                )
222                .await
223                {
224                    Ok(result) => result.map_err(|e| e.to_string()),
225                    Err(_) => Err(format!(
226                        "gave up after {}ms (connect_budget_ms)",
227                        self.connect_budget.as_millis()
228                    )),
229                }
230            })
231            .await
232            .map_err(|e| format!("store '{}': connect: {}", self.name, e))?;
233        Ok(StoreConn {
234            inner: manager.clone(),
235            budget: self.connect_budget,
236            name: self.name.clone(),
237        })
238    }
239
240    /// `PING` + server version, for the Admin API connectivity check.
241    pub async fn ping(&self) -> Result<PingInfo, String> {
242        let mut conn = self.conn().await?;
243        let start = std::time::Instant::now();
244        let pong: String = redis::cmd("PING")
245            .query_async(&mut conn)
246            .await
247            .map_err(|e| format!("store '{}': ping: {}", self.name, e))?;
248        if pong != "PONG" {
249            return Err(format!(
250                "store '{}': unexpected PING reply '{}'",
251                self.name, pong
252            ));
253        }
254        let latency_ms = start.elapsed().as_millis() as u64;
255        let info: String = redis::cmd("INFO")
256            .arg("server")
257            .query_async(&mut conn)
258            .await
259            .unwrap_or_default();
260        let version = info
261            .lines()
262            .find_map(|l| {
263                l.strip_prefix("valkey_version:")
264                    .or_else(|| l.strip_prefix("redis_version:"))
265            })
266            .unwrap_or("unknown")
267            .trim()
268            .to_string();
269        Ok(PingInfo {
270            latency_ms,
271            version,
272        })
273    }
274
275    // Only this module's own tests call this — not yet used by any
276    // production code path.
277    #[allow(dead_code)]
278    pub fn name(&self) -> &str {
279        &self.name
280    }
281
282    pub fn key_prefix(&self) -> &str {
283        &self.key_prefix
284    }
285
286    pub fn fingerprint(&self) -> &str {
287        &self.fingerprint
288    }
289
290    pub fn connect_timeout(&self) -> Duration {
291        self.connect_timeout
292    }
293}
294
295#[cfg(test)]
296mod tests {
297    use super::*;
298
299    fn cfg(yaml: &str) -> StoreConfig {
300        serde_yaml::from_str(yaml).unwrap()
301    }
302
303    /// Env placeholders resolve at build; the config object stays raw; a
304    /// changed env var changes the fingerprint.
305    #[test]
306    fn test_build_resolves_env_and_fingerprints() {
307        std::env::set_var("STORE_T3_URL", "redis://127.0.0.1:6399");
308        let c = cfg("name: s1\ntype: redis\nurl: ${STORE_T3_URL}\n");
309        let built = RedisStoreClient::build(&c).unwrap();
310        assert_eq!(built.name(), "s1");
311        assert_eq!(built.key_prefix(), "fb");
312        // Raw config untouched.
313        assert_eq!(c.url, "${STORE_T3_URL}");
314
315        let fp1 = RedisStoreClient::fingerprint_of(&c);
316        std::env::set_var("STORE_T3_URL", "redis://127.0.0.1:6400");
317        let fp2 = RedisStoreClient::fingerprint_of(&c);
318        assert_ne!(fp1, fp2, "resolved env change must change the fingerprint");
319        std::env::remove_var("STORE_T3_URL");
320    }
321
322    #[test]
323    fn test_build_rejects_bad_url() {
324        let c = cfg("name: s1\ntype: redis\nurl: 'not a url'\n");
325        let err = RedisStoreClient::build(&c).unwrap_err();
326        assert!(err.contains("store 's1'"), "{err}");
327    }
328
329    /// Live-backend test; skipped unless FEATHERBIT_TEST_REDIS_URL is set
330    /// (e.g. `docker run -p 6379:6379 redis:7` then
331    /// FEATHERBIT_TEST_REDIS_URL=redis://127.0.0.1:6379 cargo test).
332    #[tokio::test]
333    async fn test_ping_live() {
334        let Ok(url) = std::env::var("FEATHERBIT_TEST_REDIS_URL") else {
335            eprintln!("skipping test_ping_live: FEATHERBIT_TEST_REDIS_URL not set");
336            return;
337        };
338        let c = cfg(&format!("name: live\ntype: redis\nurl: {url}\n"));
339        let client = RedisStoreClient::build(&c).unwrap();
340        let info = client.ping().await.unwrap();
341        assert!(!info.version.is_empty());
342    }
343
344    /// A store that cannot be reached must give up inside its budget rather
345    /// than working through the connection manager's retry schedule.
346    ///
347    /// Nothing listens on port 1, so every attempt is refused immediately and
348    /// the elapsed time is almost entirely the crate's exponential backoff:
349    /// with the defaults that is 100+200+400+800+1600+3200ms of waiting
350    /// between six attempts. The budget has to cut that short, because this
351    /// happens on the first request after an outage begins -- exactly when a
352    /// gateway should shed load fastest, not hold the request open.
353    #[tokio::test]
354    async fn test_connect_gives_up_inside_its_budget() {
355        let c = cfg("name: s1
356type: redis
357url: redis://127.0.0.1:1
358connect_budget_ms: 300
359");
360        let client = RedisStoreClient::build(&c).unwrap();
361
362        let started = std::time::Instant::now();
363        // `ConnectionManager` has no `Debug`, so `unwrap_err()` is unavailable.
364        let err = match client.conn().await {
365            Ok(_) => panic!("connecting to a closed port must fail"),
366            Err(e) => e,
367        };
368        let elapsed = started.elapsed();
369
370        assert!(
371            elapsed < Duration::from_millis(2000),
372            "connect must abandon inside its budget, not run the full retry schedule: took {elapsed:?}"
373        );
374        assert!(err.contains("store 's1'"), "{err}");
375    }
376
377    /// A failed connect must not be cached: the `OnceCell` stays uninitialised
378    /// so the next request tries again, rather than a single outage poisoning
379    /// the store for the process's lifetime.
380    #[tokio::test]
381    async fn test_a_failed_connect_is_not_cached() {
382        let c = cfg("name: s1
383type: redis
384url: redis://127.0.0.1:1
385connect_budget_ms: 300
386");
387        let client = RedisStoreClient::build(&c).unwrap();
388
389        assert!(client.conn().await.is_err());
390        let started = std::time::Instant::now();
391        assert!(client.conn().await.is_err());
392        assert!(
393            started.elapsed() > Duration::from_millis(50),
394            "a second attempt must actually retry, not return a cached failure instantly"
395        );
396    }
397
398    /// The budget has a default, so an existing config with no new key is
399    /// still bounded rather than inheriting the crate's unbounded schedule.
400    #[test]
401    fn test_connect_budget_defaults() {
402        let c = cfg("name: s1
403type: redis
404url: redis://127.0.0.1:6379
405");
406        assert_eq!(c.connect_budget_ms, 5000);
407    }
408
409    /// A TCP relay in front of the live redis that the test can kill, so the
410    /// gateway's connection sees the store vanish mid-life -- the case
411    /// `test_connect_gives_up_inside_its_budget` cannot reach, because it
412    /// never gets a connection in the first place.
413    struct KillableProxy {
414        port: u16,
415        relays: std::sync::Arc<std::sync::Mutex<Vec<tokio::task::JoinHandle<()>>>>,
416        accept: tokio::task::JoinHandle<()>,
417    }
418
419    impl KillableProxy {
420        async fn start(target: String) -> Self {
421            let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
422            let port = listener.local_addr().unwrap().port();
423            let relays = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
424            let relays_for_accept = relays.clone();
425            let accept = tokio::spawn(async move {
426                loop {
427                    let Ok((mut inbound, _)) = listener.accept().await else {
428                        return;
429                    };
430                    let target = target.clone();
431                    let task = tokio::spawn(async move {
432                        let Ok(mut outbound) = tokio::net::TcpStream::connect(target).await else {
433                            return;
434                        };
435                        let _ = tokio::io::copy_bidirectional(&mut inbound, &mut outbound).await;
436                    });
437                    relays_for_accept.lock().unwrap().push(task);
438                }
439            });
440            Self {
441                port,
442                relays,
443                accept,
444            }
445        }
446
447        /// Stops accepting and aborts every relay, which drops both ends of
448        /// each relayed socket: from the client's side the store has gone
449        /// away and its port now refuses connections.
450        fn kill(self) {
451            self.accept.abort();
452            for t in self.relays.lock().unwrap().drain(..) {
453                t.abort();
454            }
455        }
456    }
457
458    /// The budget must bound every store operation, not just the first
459    /// connection. After an outage the connection manager reconnects with
460    /// six exponentially backed-off attempts; a command issued meanwhile
461    /// awaits that whole schedule -- measured at ~55s against a stopped
462    /// container -- long after `connect_budget_ms` says the store should
463    /// have been given up on.
464    ///
465    /// Live-backend test; skipped unless FEATHERBIT_TEST_REDIS_URL is set.
466    #[tokio::test]
467    async fn test_a_command_during_reconnect_gives_up_inside_its_budget() {
468        let Ok(url) = std::env::var("FEATHERBIT_TEST_REDIS_URL") else {
469            eprintln!("skipping: FEATHERBIT_TEST_REDIS_URL not set");
470            return;
471        };
472        let target = url
473            .trim_start_matches("redis://")
474            .trim_end_matches('/')
475            .to_string();
476        let proxy = KillableProxy::start(target).await;
477        let port = proxy.port;
478
479        let c = cfg(&format!(
480            "name: s1
481type: redis
482url: redis://127.0.0.1:{port}
483connect_timeout_ms: 200
484connect_budget_ms: 300
485"
486        ));
487        let client = RedisStoreClient::build(&c).unwrap();
488        let mut conn = client.conn().await.expect("connect through the relay");
489        let pong: String = redis::cmd("PING").query_async(&mut conn).await.unwrap();
490        assert_eq!(pong, "PONG");
491
492        proxy.kill();
493
494        // The first command after the outage fails fast on the dead socket
495        // and starts the reconnect; the second is the one that used to wait
496        // out the manager's entire retry schedule.
497        let _ = redis::cmd("PING")
498            .query_async::<String>(&mut conn)
499            .await
500            .expect_err("the socket is gone");
501        let started = std::time::Instant::now();
502        let err = redis::cmd("PING")
503            .query_async::<String>(&mut conn)
504            .await
505            .expect_err("nothing listens on the relay port any more");
506        let elapsed = started.elapsed();
507
508        assert!(
509            elapsed < Duration::from_millis(1500),
510            "a command during reconnect must give up inside connect_budget_ms, not wait out the retry schedule: took {elapsed:?} ({err})"
511        );
512    }
513}
featherbit/stores/redis_store.rs

featherbit/stores/
redis_store.rs
