From d54a4bb847585ee24629ebe4394c889ceb198a19 Mon Sep 17 00:00:00 2001 From: ttatsato Date: Sat, 25 Apr 2026 10:33:24 +0900 Subject: [PATCH 1/3] =?UTF-8?q?change:=20redis=20connection=20pool=20?= =?UTF-8?q?=E2=86=92=20mutilpledConnection=E3=81=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/adapters/auth_cache/valkey.rs | 49 ++++++++++++++--------------- src/adapters/rate_limiter/valkey.rs | 18 +++-------- 2 files changed, 29 insertions(+), 38 deletions(-) diff --git a/src/adapters/auth_cache/valkey.rs b/src/adapters/auth_cache/valkey.rs index c3385ab..61eff50 100644 --- a/src/adapters/auth_cache/valkey.rs +++ b/src/adapters/auth_cache/valkey.rs @@ -1,30 +1,30 @@ use crate::adapters::auth_cache::AuthCache; use async_trait::async_trait; +use redis::aio::MultiplexedConnection; pub struct ValkeyAuthCache { - client: redis::Client, + conn: MultiplexedConnection, } impl ValkeyAuthCache { pub async fn new(url: &str) -> Result { let client = redis::Client::open(url)?; + let mut conn = client.get_multiplexed_async_connection().await?; - if let Ok(mut conn) = client.get_multiplexed_async_connection().await { - redis::cmd("PING") - .query_async::(&mut conn) - .await - .map_err(|e| eprintln!("{}", e)) - .ok(); - } + redis::cmd("PING") + .query_async::(&mut conn) + .await + .map_err(|e| eprintln!("{}", e)) + .ok(); - Ok(Self { client }) + Ok(Self { conn }) } } #[async_trait] impl AuthCache for ValkeyAuthCache { async fn get(&self, key_hash: &str) -> Option { - let mut conn = self.client.get_multiplexed_async_connection().await.ok()?; + let mut conn = self.conn.clone(); redis::cmd("GET") .arg(key_hash) .query_async::(&mut conn) @@ -33,24 +33,23 @@ impl AuthCache for ValkeyAuthCache { } async fn set(&self, key_hash: &str, value: &str, ttl_secs: u64) { - if let Ok(mut conn) = self.client.get_multiplexed_async_connection().await { - redis::cmd("SET") - .arg(key_hash) - .arg(value) - .arg("EX") - .arg(ttl_secs) - .query_async::(&mut conn) - .await - .ok(); - } + let mut conn = self.conn.clone(); + redis::cmd("SET") + .arg(key_hash) + .arg(value) + .arg("EX") + .arg(ttl_secs) + .query_async::(&mut conn) + .await + .ok(); } async fn delete(&self, key_hash: &str) { - if let Ok(mut conn) = self.client.get_multiplexed_async_connection().await - && let Err(e) = redis::cmd("DEL") - .arg(key_hash) - .query_async::(&mut conn) - .await + let mut conn = self.conn.clone(); + if let Err(e) = redis::cmd("DEL") + .arg(key_hash) + .query_async::(&mut conn) + .await { eprintln!("failed to delete auth cache key {}: {}", key_hash, e); } diff --git a/src/adapters/rate_limiter/valkey.rs b/src/adapters/rate_limiter/valkey.rs index cf0030a..1fa0627 100644 --- a/src/adapters/rate_limiter/valkey.rs +++ b/src/adapters/rate_limiter/valkey.rs @@ -2,10 +2,11 @@ use super::{RateLimit, RateLimitPeriod, RateLimiter, RateLimiterError}; use async_trait::async_trait; use chrono::{Datelike, Utc}; use redis::AsyncCommands; +use redis::aio::MultiplexedConnection; use uuid::Uuid; pub struct ValkeyRateLimiter { - client: redis::Client, + conn: MultiplexedConnection, } impl ValkeyRateLimiter { @@ -13,7 +14,6 @@ impl ValkeyRateLimiter { let client = redis::Client::open(url).map_err(|e| RateLimiterError::Internal(e.to_string()))?; - // 起動時に接続確認 let mut conn = client .get_multiplexed_async_connection() .await @@ -23,7 +23,7 @@ impl ValkeyRateLimiter { .await .map_err(|e| RateLimiterError::Internal(e.to_string()))?; - Ok(Self { client }) + Ok(Self { conn }) } fn tokens_key(&self, consumer_id: Uuid, period: &RateLimitPeriod) -> String { @@ -157,11 +157,7 @@ impl RateLimiter for ValkeyRateLimiter { consumer_id: Uuid, limits: &[RateLimit], ) -> Result { - let mut conn = self - .client - .get_multiplexed_async_connection() - .await - .map_err(|e| RateLimiterError::Internal(e.to_string()))?; + let mut conn = self.conn.clone(); let now = Utc::now().timestamp_millis() as f64 / 1000.0; @@ -198,11 +194,7 @@ impl RateLimiter for ValkeyRateLimiter { period: &RateLimitPeriod, max_requests: i64, ) -> Result { - let mut conn = self - .client - .get_multiplexed_async_connection() - .await - .map_err(|e| RateLimiterError::Internal(e.to_string()))?; + let mut conn = self.conn.clone(); let tokens_key = self.tokens_key(consumer_id, period); let last_key = self.last_key(consumer_id, period); From 6a3a3ad2fe735715a2f18520fccd31a0ad585b8d Mon Sep 17 00:00:00 2001 From: ttatsato Date: Thu, 30 Apr 2026 09:42:43 +0900 Subject: [PATCH 2/3] add(ADR): multiplexed connection for valkey --- ...share-multiplexed-connection-for-valkey.md | 59 +++++++++++++++++++ 1 file changed, 59 insertions(+) create mode 100644 docs/adr/0002-share-multiplexed-connection-for-valkey.md diff --git a/docs/adr/0002-share-multiplexed-connection-for-valkey.md b/docs/adr/0002-share-multiplexed-connection-for-valkey.md new file mode 100644 index 0000000..d7516db --- /dev/null +++ b/docs/adr/0002-share-multiplexed-connection-for-valkey.md @@ -0,0 +1,59 @@ +# 0002. Share a MultiplexedConnection across Valkey adapter calls + +# Status +accepted + +# Context + +`ValkeyAuthCache` and `ValkeyRateLimiter` previously held a `redis::Client` and +called `get_multiplexed_async_connection()` on every operation. That added +connection setup overhead to each cache lookup / rate-limit check, and made the +number of active Valkey connections at any given moment hard to reason about. + +This is the same pattern we applied to the HTTP side in PR #70 (sharing one +`reqwest::Client` across proxy requests instead of constructing one per request). +Valkey adapters were the next obvious target. + +# Decision + +Both Valkey adapters now hold a `redis::aio::MultiplexedConnection` directly, +established once in `new()`: + +- `ValkeyAuthCache` and `ValkeyRateLimiter` store `conn: MultiplexedConnection` +- The startup `PING` runs on the same connection that will be reused at runtime +- Each operation does `let mut conn = self.conn.clone();` — `MultiplexedConnection` + is designed to be cheaply cloned and used concurrently from many tasks, so + this is the intended usage pattern from the `redis` crate + +# Alternatives Considered + +- **Status quo: `get_multiplexed_async_connection()` per call.** Rejected: + redundant setup work on every request; connection count is unpredictable. +- **Use a pool (bb8-redis / deadpool-redis).** Rejected for now: a single + `MultiplexedConnection` is already safe to share across concurrent tasks, so + a pool adds dependencies and complexity for little gain at the current scale. + Worth revisiting if we ever need per-tenant isolation, bulkheads between + high-volume callers, or tighter control over reconnection behavior. +- **Spawn / hold one connection per worker manually.** Rejected: defeats the + purpose of a multiplexed connection, which already pipelines commands from + many callers over a single underlying socket. + +# Consequences + +- Positive: + - Eliminates per-operation connection setup on the hot path + - Code is shorter and has fewer error branches (no per-call connection failure case) + - Connection count per adapter instance is fixed and predictable +- Negative / trade-offs: + - All callers share one underlying socket; if the connection is unhealthy, + everyone is affected until the `redis` crate's reconnection logic recovers. + No bulkhead between high-volume and low-volume callers. + - If we later need pooling, we have to revisit both adapters. +- Affected modules / operations: + - `src/adapters/auth_cache/valkey.rs` + - `src/adapters/rate_limiter/valkey.rs` + +# References + +- [PR #71](https://github.com/ttatsato/usage-gate/pull/71) — this change +- [PR #70](https://github.com/ttatsato/usage-gate/pull/70) — same pattern applied to the shared `reqwest::Client` From 1d280f2164f44ba478d13cb549d1de036ef3d922 Mon Sep 17 00:00:00 2001 From: ttatsato Date: Thu, 30 Apr 2026 09:57:11 +0900 Subject: [PATCH 3/3] =?UTF-8?q?change:=20MultiplexedConnection=20=E2=86=92?= =?UTF-8?q?=20ConnectionManager?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Cargo.lock | 21 ++++ Cargo.toml | 2 +- ...share-multiplexed-connection-for-valkey.md | 104 +++++++++++++----- src/adapters/auth_cache/valkey.rs | 6 +- src/adapters/rate_limiter/valkey.rs | 7 +- 5 files changed, 102 insertions(+), 38 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 1cab9d3..2beb1c4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -32,6 +32,15 @@ version = "1.0.102" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c" +[[package]] +name = "arc-swap" +version = "1.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6a3a1fd6f75306b68087b831f025c712524bcb19aad54e557b1129cfa0a2b207" +dependencies = [ + "rustversion", +] + [[package]] name = "arcstr" version = "1.2.0" @@ -133,6 +142,15 @@ dependencies = [ "tracing", ] +[[package]] +name = "backon" +version = "1.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cffb0e931875b666fc4fcb20fee52e9bbd1ef836fd9e9e04ec21555f9f85f7ef" +dependencies = [ + "fastrand", +] + [[package]] name = "base64" version = "0.22.1" @@ -1410,11 +1428,14 @@ version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f44e94c96d8870a387d88ce3de3fdd608cbfc0705f03cb343cdde91509d3e49a" dependencies = [ + "arc-swap", "arcstr", "async-lock", + "backon", "bytes", "cfg-if", "combine", + "futures-channel", "futures-util", "itoa", "num-bigint", diff --git a/Cargo.toml b/Cargo.toml index 9ff250c..9a3d74a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -18,7 +18,7 @@ chrono = { version = "0.4", features = ["serde", "clock"] } sha2="0.10" hex="0.4" reqwest = { version = "0.12", features = ["json", "stream"] } -redis = { version = "1.2", features = ["tokio-comp"] } +redis = { version = "1.2", features = ["tokio-comp", "connection-manager"] } http-body-util = "0.1" [dev-dependencies] diff --git a/docs/adr/0002-share-multiplexed-connection-for-valkey.md b/docs/adr/0002-share-multiplexed-connection-for-valkey.md index d7516db..102810f 100644 --- a/docs/adr/0002-share-multiplexed-connection-for-valkey.md +++ b/docs/adr/0002-share-multiplexed-connection-for-valkey.md @@ -1,59 +1,103 @@ -# 0002. Share a MultiplexedConnection across Valkey adapter calls +# 0002. Share a ConnectionManager across Valkey adapter calls # Status accepted # Context -`ValkeyAuthCache` and `ValkeyRateLimiter` previously held a `redis::Client` and -called `get_multiplexed_async_connection()` on every operation. That added -connection setup overhead to each cache lookup / rate-limit check, and made the +`ValkeyAuthCache` and `ValkeyRateLimiter` originally held a `redis::Client` and called +`get_multiplexed_async_connection()` on every operation. That created a fresh connection +per cache lookup / rate-limit check, added setup overhead to the hot path, and made the number of active Valkey connections at any given moment hard to reason about. -This is the same pattern we applied to the HTTP side in PR #70 (sharing one -`reqwest::Client` across proxy requests instead of constructing one per request). +This is the same shape of problem we addressed for HTTP in PR #70 (sharing one +`reqwest::Client` across proxy requests instead of constructing one per request). The Valkey adapters were the next obvious target. # Decision -Both Valkey adapters now hold a `redis::aio::MultiplexedConnection` directly, -established once in `new()`: +Both Valkey adapters now hold a `redis::aio::ConnectionManager`, established once in `new()`: -- `ValkeyAuthCache` and `ValkeyRateLimiter` store `conn: MultiplexedConnection` +- `ValkeyAuthCache` and `ValkeyRateLimiter` store `conn: ConnectionManager` +- `ConnectionManager::new(client)` is called at startup; the same handle is reused for + the lifetime of the process +- Each operation does `let mut conn = self.conn.clone();` — `ConnectionManager` is + cheaply cloneable (Arc-backed) and is designed to be shared across many concurrent tasks - The startup `PING` runs on the same connection that will be reused at runtime -- Each operation does `let mut conn = self.conn.clone();` — `MultiplexedConnection` - is designed to be cheaply cloned and used concurrently from many tasks, so - this is the intended usage pattern from the `redis` crate + +`ConnectionManager` is a `MultiplexedConnection`-based handle from the `redis` crate +that adds **automatic reconnection** on transient failures. Memory profile and +per-request behavior are the same as a bare `MultiplexedConnection` (one underlying +socket, pipelined commands), but if the connection drops, the manager re-establishes +it in the background instead of leaving callers with a permanently dead handle. + +Adds the `connection-manager` feature to the `redis` crate in `Cargo.toml`. # Alternatives Considered -- **Status quo: `get_multiplexed_async_connection()` per call.** Rejected: - redundant setup work on every request; connection count is unpredictable. -- **Use a pool (bb8-redis / deadpool-redis).** Rejected for now: a single - `MultiplexedConnection` is already safe to share across concurrent tasks, so - a pool adds dependencies and complexity for little gain at the current scale. - Worth revisiting if we ever need per-tenant isolation, bulkheads between - high-volume callers, or tighter control over reconnection behavior. -- **Spawn / hold one connection per worker manually.** Rejected: defeats the - purpose of a multiplexed connection, which already pipelines commands from - many callers over a single underlying socket. +- **Status quo: `get_multiplexed_async_connection()` per call.** Rejected: redundant + setup work on every request; connection count is unpredictable; nothing about it + was an intentional design choice. +- **Bare `MultiplexedConnection` (no auto-reconnect).** Initially proposed in PR #71. + Rejected after review (Codex flagged this as P1/P2): the `redis` crate's + `MultiplexedConnection` does **not** reconnect on failure, so a Valkey restart or + TCP drop leaves all cloned handles permanently failing. For an authenticated request + path, that turns a transient Valkey blip into a continuous HTTP 500 stream until the + API Gateway process is restarted. `ConnectionManager` solves this without changing + the per-request cost or memory profile. +- **Traditional connection pool (`bb8-redis` / `deadpool-redis`).** Rejected: these + pools maintain N independent connections with lease/return semantics, which adds + bookkeeping `ConnectionManager` doesn't need (each `ConnectionManager` is already + safe to share across many concurrent tasks). The N-socket cost negates the memory + advantage that makes a shared connection attractive for an API Gateway, with no + matching benefit at our current scale. +- **Sharded `ConnectionManager` (`Vec` with round-robin / + random selection).** Not yet — kept as a future option. This is closer to + "sharding" than "pooling": no lease/return needed, just pick one of N handles per + request. Trade-offs: + - Pros: relieves multiplexer mutex contention at very high concurrency, exploits + Valkey 8's IO threads on the server side, mitigates head-of-line blocking from + heavy Lua scripts (e.g. the rate-limiter token bucket), enables bulkheading + between workload classes (auth_cache vs rate_limiter, or per-tenant) + - Cons: linear FD / memory growth in N (small), reconnect thundering-herd risk on + Valkey restart, more complex to monitor per-connection health + - Trigger to revisit: profiling shows the single multiplexer / single socket is + the bottleneck, OR a concrete bulkheading requirement emerges (e.g. one tenant's + rate-limit traffic starving the auth cache). Adopting it speculatively in the + MVP phase is over-engineering. +- **Handle reconnection at the infrastructure layer (k8s liveness probe, HA Valkey).** + Rejected as a substitute, kept as defense-in-depth. Liveness-probe-based restart + converts every Valkey hiccup into a full pod restart (during which all requests + fail); HA Valkey reduces failure frequency but failover still produces TCP drops + the client must handle. Application-level reconnection is the right primary + mechanism; infra HA is complementary, not a replacement. +- **Hold one connection per worker manually.** Rejected: defeats the purpose of a + multiplexed connection, which already pipelines commands from many callers over a + single underlying socket. # Consequences - Positive: - Eliminates per-operation connection setup on the hot path - - Code is shorter and has fewer error branches (no per-call connection failure case) - - Connection count per adapter instance is fixed and predictable + - Auto-reconnects on transient Valkey failures — transient outages no longer require + a process restart + - Code is shorter and has fewer error branches than the original per-call setup + - Connection count per adapter instance is fixed and predictable (one socket each) - Negative / trade-offs: - - All callers share one underlying socket; if the connection is unhealthy, - everyone is affected until the `redis` crate's reconnection logic recovers. - No bulkhead between high-volume and low-volume callers. - - If we later need pooling, we have to revisit both adapters. + - All callers share one underlying socket per adapter; no bulkhead between + high-volume and low-volume callers + - Reconnection is implicit — failures are retried behind the application's back, + which can hide a degraded Valkey from monitoring unless we instrument it explicitly + - Adds the `connection-manager` feature to the `redis` crate (negligible build cost) - Affected modules / operations: - `src/adapters/auth_cache/valkey.rs` - `src/adapters/rate_limiter/valkey.rs` + - `Cargo.toml` (added `connection-manager` feature) # References -- [PR #71](https://github.com/ttatsato/usage-gate/pull/71) — this change -- [PR #70](https://github.com/ttatsato/usage-gate/pull/70) — same pattern applied to the shared `reqwest::Client` +- [PR #71](https://github.com/ttatsato/usage-gate/pull/71) — this change, and the + Codex review thread that prompted switching from bare `MultiplexedConnection` to + `ConnectionManager` +- [PR #70](https://github.com/ttatsato/usage-gate/pull/70) — same shared-handle + pattern applied to `reqwest::Client` diff --git a/src/adapters/auth_cache/valkey.rs b/src/adapters/auth_cache/valkey.rs index 61eff50..77174bd 100644 --- a/src/adapters/auth_cache/valkey.rs +++ b/src/adapters/auth_cache/valkey.rs @@ -1,15 +1,15 @@ use crate::adapters::auth_cache::AuthCache; use async_trait::async_trait; -use redis::aio::MultiplexedConnection; +use redis::aio::ConnectionManager; pub struct ValkeyAuthCache { - conn: MultiplexedConnection, + conn: ConnectionManager, } impl ValkeyAuthCache { pub async fn new(url: &str) -> Result { let client = redis::Client::open(url)?; - let mut conn = client.get_multiplexed_async_connection().await?; + let mut conn = ConnectionManager::new(client).await?; redis::cmd("PING") .query_async::(&mut conn) diff --git a/src/adapters/rate_limiter/valkey.rs b/src/adapters/rate_limiter/valkey.rs index 1fa0627..3ec6d7c 100644 --- a/src/adapters/rate_limiter/valkey.rs +++ b/src/adapters/rate_limiter/valkey.rs @@ -2,11 +2,11 @@ use super::{RateLimit, RateLimitPeriod, RateLimiter, RateLimiterError}; use async_trait::async_trait; use chrono::{Datelike, Utc}; use redis::AsyncCommands; -use redis::aio::MultiplexedConnection; +use redis::aio::ConnectionManager; use uuid::Uuid; pub struct ValkeyRateLimiter { - conn: MultiplexedConnection, + conn: ConnectionManager, } impl ValkeyRateLimiter { @@ -14,8 +14,7 @@ impl ValkeyRateLimiter { let client = redis::Client::open(url).map_err(|e| RateLimiterError::Internal(e.to_string()))?; - let mut conn = client - .get_multiplexed_async_connection() + let mut conn = ConnectionManager::new(client) .await .map_err(|e| RateLimiterError::Internal(e.to_string()))?; redis::cmd("PING")