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 new file mode 100644 index 0000000..102810f --- /dev/null +++ b/docs/adr/0002-share-multiplexed-connection-for-valkey.md @@ -0,0 +1,103 @@ +# 0002. Share a ConnectionManager across Valkey adapter calls + +# Status +accepted + +# Context + +`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 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::ConnectionManager`, established once in `new()`: + +- `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 + +`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; 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 + - 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 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, 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 c3385ab..77174bd 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::ConnectionManager; pub struct ValkeyAuthCache { - client: redis::Client, + conn: ConnectionManager, } impl ValkeyAuthCache { pub async fn new(url: &str) -> Result { let client = redis::Client::open(url)?; + let mut conn = ConnectionManager::new(client).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..3ec6d7c 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::ConnectionManager; use uuid::Uuid; pub struct ValkeyRateLimiter { - client: redis::Client, + conn: ConnectionManager, } impl ValkeyRateLimiter { @@ -13,9 +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") @@ -23,7 +22,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 +156,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 +193,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);