diff --git a/.env.example b/.env.example index e671c4668d..c364f4d3df 100644 --- a/.env.example +++ b/.env.example @@ -31,9 +31,6 @@ PGDATABASE=buzz # Redis 7 # ----------------------------------------------------------------------------- REDIS_URL=redis://localhost:6379 -# Max connections in the relay's shared Redis pool (default 16). -# BUZZ_REDIS_POOL_SIZE=16 - # Max connections in each of the relay's Postgres pools — writer and, when # READ_DATABASE_URL is set, reader (default 50). # BUZZ_DB_POOL_SIZE=50 diff --git a/.github/workflows/_ci-relay.yml b/.github/workflows/_ci-relay.yml index bf57fb424b..7652bab800 100644 --- a/.github/workflows/_ci-relay.yml +++ b/.github/workflows/_ci-relay.yml @@ -122,6 +122,7 @@ jobs: --cargo-profile ci \ -p buzz-db \ -p buzz-relay \ + -p buzz-relay-mesh \ -p buzz-test-client \ --lib \ --test e2e_event_reminder \ @@ -574,6 +575,14 @@ jobs: --archive-file target/ci/backend-integration-tests.tar.zst \ -E 'package(buzz-relay) and test(/^api::nip_fi::route_integration_tests::external_infra_redis::/)' \ --run-ignored ignored-only + - name: Mesh ready-registry index regressions + env: + BUZZ_TEST_REDIS_URL: redis://localhost:6379 + run: | + cargo nextest run \ + --archive-file target/ci/backend-integration-tests.tar.zst \ + -E 'package(buzz-relay-mesh) and test(/^registry::tests::external_infra_redis::/)' \ + --run-ignored ignored-only - name: Workflow message provenance unit tests # The relay's workflow_sink suite is not selected by the infra-free # unit job. Its ignored database cases run in the isolated PostgreSQL diff --git a/Cargo.lock b/Cargo.lock index dfef262b3b..9e5951c043 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -979,7 +979,6 @@ dependencies = [ "buzz-search", "buzz-workflow", "clap", - "deadpool-redis", "hex", "nostr 0.44.7", "rustls", @@ -1212,9 +1211,9 @@ dependencies = [ "buzz-core", "buzz-db", "buzz-media", + "buzz-pubsub", "chrono", "clap", - "deadpool-redis", "hex", "nostr 0.44.7", "redis", @@ -1369,10 +1368,10 @@ dependencies = [ name = "buzz-pubsub" version = "0.1.0" dependencies = [ + "arc-swap", "buzz-auth", "buzz-core", "chrono", - "deadpool-redis", "futures-util", "nostr 0.44.7", "redis", @@ -1449,7 +1448,6 @@ dependencies = [ "bytes", "chrono", "dashmap", - "deadpool-redis", "flate2", "futures", "futures-util", @@ -1502,8 +1500,8 @@ dependencies = [ name = "buzz-relay-mesh" version = "0.1.0" dependencies = [ + "buzz-pubsub", "bytes", - "deadpool-redis", "futures-util", "hex", "hmac 0.13.0", @@ -2608,36 +2606,6 @@ dependencies = [ "syn 2.0.117", ] -[[package]] -name = "deadpool" -version = "0.13.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "883466cb8db62725aee5f4a6011e8a5d42912b42632df32aad57fc91127c6e04" -dependencies = [ - "deadpool-runtime", - "num_cpus", - "tokio", -] - -[[package]] -name = "deadpool-redis" -version = "0.23.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bafa30c49dafe086d10116074e422ad7fc1c3cf554697e744a3ab112599ebd09" -dependencies = [ - "deadpool", - "redis", -] - -[[package]] -name = "deadpool-runtime" -version = "0.3.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2657f61fb1dd8bf37a8d51093cc7cee4e77125b22f7753f49b289f831bec2bae" -dependencies = [ - "tokio", -] - [[package]] name = "defmt" version = "1.1.1" @@ -6709,16 +6677,6 @@ dependencies = [ "autocfg", ] -[[package]] -name = "num_cpus" -version = "1.17.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "91df4bbde75afed763b708b7eee1e8e7651e02d97f6d5dd763e89367e957b23b" -dependencies = [ - "hermit-abi", - "libc", -] - [[package]] name = "num_enum" version = "0.7.6" @@ -8425,14 +8383,11 @@ version = "1.2.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bae41a63fd0b8a5372f82b21e810e09a316f5dd7efd96bf08e678fb240fc1918" dependencies = [ - "arc-swap", "arcstr", "async-lock", - "backon", "bytes", "cfg-if 1.0.4", "combine", - "futures-channel", "futures-util", "itoa", "num-bigint", diff --git a/Cargo.toml b/Cargo.toml index 6207d5a8c0..25a39ea4c7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -62,8 +62,7 @@ sqlx = { version = "0.9", features = [ ] } # Redis -redis = { version = "1.0", features = ["tokio-comp", "connection-manager", "tokio-rustls-comp"] } -deadpool-redis = { version = "0.23", features = ["rt_tokio_1"] } +redis = { version = "1.0", features = ["tokio-comp", "tokio-rustls-comp"] } # Kubernetes (buzz-backend-kubernetes provider). No `ring` feature here: the # process-level CryptoProvider is installed explicitly at startup, matching diff --git a/Justfile b/Justfile index 86c2f4d05b..48f1079a97 100644 --- a/Justfile +++ b/Justfile @@ -407,8 +407,9 @@ test-unit: cargo nextest run -p buzz-auth --test mesh_jwks_proxy cargo nextest run -p buzz-audit --lib # Cross-pod disconnect payload tests (NIP-FI and community archive - # fence); all infra-free. - cargo nextest run -p buzz-pubsub --lib -E 'test(/^conn_control::tests::/)' + # fence) and the shared Redis connection's timeout bounds (loopback + # fake peers); all infra-free. Keep the Cargo fallback in step. + cargo nextest run -p buzz-pubsub --lib -E 'test(/^conn_control::tests::/) + test(/^connection::tests::/)' # buzz-auth NIP-FI verifier doctests. The sealed-authority # `compile_fail` doctests prove the default-feature public API alone # cannot forge the issuer→JWKS authority; nextest does not run diff --git a/crates/buzz-admin/Cargo.toml b/crates/buzz-admin/Cargo.toml index d8d075a8b2..10354c2f92 100644 --- a/crates/buzz-admin/Cargo.toml +++ b/crates/buzz-admin/Cargo.toml @@ -27,7 +27,6 @@ serde_json = { workspace = true } serde = { workspace = true } anyhow = { workspace = true } hex = { workspace = true } -deadpool-redis = { workspace = true } # Redis TLS (rediss://, e.g. ElastiCache) uses rustls, which needs a process # CryptoProvider installed at startup. The workspace redis TLS feature compiles # both aws-lc-rs and ring in transitively, so rustls can't auto-select one — we diff --git a/crates/buzz-admin/src/communities.rs b/crates/buzz-admin/src/communities.rs index 4830d7d9ef..fe1fa3559a 100644 --- a/crates/buzz-admin/src/communities.rs +++ b/crates/buzz-admin/src/communities.rs @@ -152,9 +152,8 @@ async fn unarchive( async fn publish_disconnect(tenant: &TenantContext, command: &ConnControl) -> Result { let redis_url = std::env::var("REDIS_URL").context("REDIS_URL is required")?; - let redis_pool = deadpool_redis::Config::from_url(&redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .context("Redis pool creation failed")?; + let redis_pool = + buzz_pubsub::redis_connection(&redis_url).context("Redis connection setup failed")?; let pubsub = PubSubManager::new(&redis_url, redis_pool) .await .context("PubSub init failed")?; diff --git a/crates/buzz-admin/src/main.rs b/crates/buzz-admin/src/main.rs index b286037ac1..36af8df338 100644 --- a/crates/buzz-admin/src/main.rs +++ b/crates/buzz-admin/src/main.rs @@ -663,11 +663,8 @@ async fn connect_member_services() -> Result<(Db, Arc, Keys)> { let redis_url = std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://localhost:6379".to_string()); - let redis_pool = { - let cfg = deadpool_redis::Config::from_url(&redis_url); - cfg.create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .map_err(|e| anyhow::anyhow!("Redis pool creation failed: {e}"))? - }; + let redis_pool = buzz_pubsub::redis_connection(&redis_url) + .map_err(|e| anyhow::anyhow!("Redis connection setup failed: {e}"))?; let pubsub = Arc::new( PubSubManager::new(&redis_url, redis_pool) diff --git a/crates/buzz-deletion/Cargo.toml b/crates/buzz-deletion/Cargo.toml index 6937974c84..bfc8cb8d24 100644 --- a/crates/buzz-deletion/Cargo.toml +++ b/crates/buzz-deletion/Cargo.toml @@ -13,9 +13,9 @@ thiserror = { workspace = true } buzz-core = { workspace = true } buzz-db = { workspace = true } buzz-media = { workspace = true } +buzz-pubsub = { workspace = true } chrono = { workspace = true } clap = { version = "4", features = ["derive"] } -deadpool-redis = { workspace = true } hex = { workspace = true } redis = { workspace = true } serde = { workspace = true } diff --git a/crates/buzz-deletion/src/lib.rs b/crates/buzz-deletion/src/lib.rs index c9e7842394..158681a96a 100644 --- a/crates/buzz-deletion/src/lib.rs +++ b/crates/buzz-deletion/src/lib.rs @@ -25,6 +25,7 @@ use buzz_media::{ is_tenant_owned_key, tenant_prefixes, BulkDeleteOutcome, MediaStorage, ObjectVersionKind, ObjectVersionRef, }; +use buzz_pubsub::{redis_connection, RedisConnection}; use clap::Subcommand; use serde::Serialize; use tokio_util::sync::CancellationToken; @@ -340,7 +341,7 @@ impl LoopMode { struct Services { store: DeletionStore, media: Arc, - redis: deadpool_redis::Pool, + redis: RedisConnection, } #[derive(Debug, thiserror::Error)] @@ -612,14 +613,7 @@ async fn connect_services_with_store(store: DeletionStore) -> Result { }; let media = Arc::new(MediaStorage::new(&media_config)?); let redis_url = required_env("REDIS_URL")?; - let mut redis_config = deadpool_redis::Config::from_url(&redis_url); - redis_config.pool = Some(deadpool_redis::PoolConfig::new(env_parse( - "BUZZ_REDIS_POOL_SIZE", - 16, - ))); - let redis = redis_config - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .context("create deletion Redis pool")?; + let redis = redis_connection(&redis_url).context("create deletion Redis connection")?; Ok(Services { store, media, @@ -1689,24 +1683,24 @@ async fn verify_storage_absence(services: &Services, request: &DeletionRequest) } async fn publish_disconnect_community( - pool: &deadpool_redis::Pool, + pool: &RedisConnection, community: buzz_core::CommunityId, ) -> Result<()> { - let mut connection = pool.get().await?; + let mut connection = pool.clone(); let channel = format!("buzz:{community}:conn-control"); let _: u64 = redis::cmd("PUBLISH") .arg(channel) .arg(r#"{"op":"DisconnectCommunity"}"#) - .query_async(&mut *connection) + .query_async(&mut connection) .await?; Ok(()) } async fn purge_redis_namespace( - pool: &deadpool_redis::Pool, + pool: &RedisConnection, community: buzz_core::CommunityId, ) -> Result { - let mut connection = pool.get().await?; + let mut connection = pool.clone(); let pattern = format!("buzz:{community}:*"); let mut cursor = 0u64; let mut deleted = 0u64; @@ -1717,12 +1711,12 @@ async fn purge_redis_namespace( .arg(&pattern) .arg("COUNT") .arg(1000) - .query_async(&mut *connection) + .query_async(&mut connection) .await?; if !keys.is_empty() { let count: u64 = redis::cmd("UNLINK") .arg(&keys) - .query_async(&mut *connection) + .query_async(&mut connection) .await?; deleted = deleted.saturating_add(count); } @@ -1740,7 +1734,7 @@ fn scan_proves_absence(pages: &[(u64, Vec)]) -> bool { } async fn scan_redis_namespace( - connection: &mut deadpool_redis::Connection, + connection: &mut RedisConnection, pattern: &str, ) -> Result)>> { let mut cursor = 0u64; @@ -1752,7 +1746,7 @@ async fn scan_redis_namespace( .arg(pattern) .arg("COUNT") .arg(1000) - .query_async(&mut **connection) + .query_async(connection) .await?; cursor = page.0; pages.push(page); @@ -1763,10 +1757,10 @@ async fn scan_redis_namespace( } async fn verify_redis_absence( - pool: &deadpool_redis::Pool, + pool: &RedisConnection, community: buzz_core::CommunityId, ) -> Result<()> { - let mut connection = pool.get().await?; + let mut connection = pool.clone(); let pattern = format!("buzz:{community}:*"); // SCAN is weakly consistent. Two complete empty passes ensure a cursor // rollover or concurrent expiry cannot make one sparse pass look absent. @@ -1957,9 +1951,8 @@ mod postgres_tests { }) .expect("construct unused media service"), ), - redis: deadpool_redis::Config::from_url("redis://127.0.0.1:1") - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("construct unused Redis pool"), + redis: redis_connection("redis://127.0.0.1:1") + .expect("construct unused Redis connection"), }; (db, services, claim) } @@ -2028,9 +2021,8 @@ mod postgres_tests { }) .expect("construct unused media service"), ), - redis: deadpool_redis::Config::from_url("redis://127.0.0.1:1") - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("construct unused Redis pool"), + redis: redis_connection("redis://127.0.0.1:1") + .expect("construct unused Redis connection"), }; (db, services, *request, inventory) } diff --git a/crates/buzz-pubsub/Cargo.toml b/crates/buzz-pubsub/Cargo.toml index 2ee2d6b47d..0d859b5e8d 100644 --- a/crates/buzz-pubsub/Cargo.toml +++ b/crates/buzz-pubsub/Cargo.toml @@ -11,7 +11,6 @@ description = "Redis pub/sub fan-out, presence, and typing indicators for Buzz" buzz-core = { workspace = true } buzz-auth = { workspace = true } redis = { workspace = true } -deadpool-redis = { workspace = true } tokio = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } @@ -21,6 +20,7 @@ tracing = { workspace = true } thiserror = { workspace = true } nostr = { workspace = true } futures-util = { workspace = true } +arc-swap = { workspace = true } [dev-dependencies] tokio = { workspace = true } diff --git a/crates/buzz-pubsub/src/connection.rs b/crates/buzz-pubsub/src/connection.rs new file mode 100644 index 0000000000..5a824eff1d --- /dev/null +++ b/crates/buzz-pubsub/src/connection.rs @@ -0,0 +1,387 @@ +//! The relay's shared Redis command connection. +//! +//! Every non-pub/sub Redis command (PUBLISH, rate-limit scripts, replay +//! guards, presence, the session directory, the mesh registry) goes through +//! one multiplexed connection. Redis pipelines concurrent commands on a single +//! socket and executes them one at a time anyway, so a pool of exclusively +//! checked-out sockets only added a queue to wait in and a PING round trip per +//! checkout. +//! +//! Recovery lives here rather than in redis-rs's `ConnectionManager`, which +//! keeps a socket that stops answering and caches a failed AUTH handshake +//! until the process restarts. This connection retires its socket on any +//! setup failure, response timeout or dropped connection, and the next command +//! dials a fresh one. A failed command is never replayed: a timed-out mutation +//! may already have run. + +use std::sync::Arc; +use std::time::Duration; + +use arc_swap::ArcSwap; +use futures_util::future::{BoxFuture, FutureExt, Shared}; +use redis::aio::{ConnectionLike, MultiplexedConnection}; +use redis::{AsyncConnectionConfig, Client, Cmd, Pipeline, RedisFuture, RedisResult, Value}; + +use crate::error::PubSubError; + +/// Bound on one connect attempt, handshake included. A command issued while +/// the connection is down waits at most this long for it before failing. +pub const REDIS_CONNECT_TIMEOUT: Duration = Duration::from_secs(1); + +/// Bound on one command's response (the redis 1.x driver default, pinned +/// here because callers budget against it). +pub const REDIS_RESPONSE_TIMEOUT: Duration = Duration::from_millis(500); + +/// Shared Redis command connection. Cloning is an `Arc` bump; every clone +/// sends on the same socket and moves to its replacement together. +#[derive(Clone)] +pub struct RedisConnection(Arc); + +struct Inner { + client: Client, + config: AsyncConnectionConfig, + /// The current socket's dial. Every command shares it; a failed or + /// retired dial is swapped for a fresh one. + socket: ArcSwap, +} + +/// One dial of the shared socket, awaited by every command that arrives while +/// it runs. Lazy: nothing connects until a command polls it. +type Dial = Shared>>; + +fn dial(client: &Client, config: &AsyncConnectionConfig) -> Arc { + let (client, config) = (client.clone(), config.clone()); + let dial = async move { + client + .get_multiplexed_async_connection_with_config(&config) + .await + }; + Arc::new(dial.boxed().shared()) +} + +/// Build the shared connection without dialing; the first command connects. +/// +/// There are no retries inside a command: a failed connect or command is that +/// command's error, and the next command dials afresh if the socket was +/// retired. So no caller waits longer than [`REDIS_CONNECT_TIMEOUT`] plus +/// [`REDIS_RESPONSE_TIMEOUT`] per command, during an outage or otherwise. +pub fn redis_connection(redis_url: &str) -> Result { + let client = Client::open(redis_url)?; + let config = AsyncConnectionConfig::new() + .set_connection_timeout(Some(REDIS_CONNECT_TIMEOUT)) + .set_response_timeout(Some(REDIS_RESPONSE_TIMEOUT)); + let socket = ArcSwap::new(dial(&client, &config)); + Ok(RedisConnection(Arc::new(Inner { + client, + config, + socket, + }))) +} + +impl RedisConnection { + async fn socket(&self) -> RedisResult<(Arc, MultiplexedConnection)> { + let current = self.0.socket.load_full(); + match (*current).clone().await { + Ok(socket) => Ok((current, socket)), + Err(err) => { + self.retire(¤t); + Err(err) + } + } + } + + /// Replace `current` unless another command already did, so one failure + /// wave costs one dial. The retired socket closes once the commands still + /// waiting on it finish or time out, which frees their reply slots. + fn retire(&self, current: &Arc) { + self.0 + .socket + .compare_and_swap(current, dial(&self.0.client, &self.0.config)); + } + + fn settle(&self, current: &Arc, result: RedisResult) -> RedisResult { + if let Err(err) = &result { + // A timeout can't tell a slow server from a socket that will never + // answer again, so it retires the socket too: on a healthy server + // that costs one handshake. + if err.is_timeout() || err.is_unrecoverable_error() { + self.retire(current); + } + } + result + } +} + +impl ConnectionLike for RedisConnection { + fn req_packed_command<'a>(&'a mut self, cmd: &'a Cmd) -> RedisFuture<'a, Value> { + async move { + let (current, mut socket) = self.socket().await?; + let result = socket.send_packed_command(cmd).await; + self.settle(¤t, result) + } + .boxed() + } + + fn req_packed_commands<'a>( + &'a mut self, + cmd: &'a Pipeline, + offset: usize, + count: usize, + ) -> RedisFuture<'a, Vec> { + async move { + let (current, mut socket) = self.socket().await?; + let result = socket.send_packed_commands(cmd, offset, count).await; + self.settle(¤t, result) + } + .boxed() + } + + fn get_db(&self) -> i64 { + self.0.client.get_connection_info().redis_settings().db() + } +} + +#[cfg(test)] +mod tests { + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::Arc; + use std::time::Instant; + + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use tokio::net::TcpListener; + + use super::*; + + /// Scheduling slack on top of a timeout before a test calls it unbounded. + const SLACK: Duration = Duration::from_millis(400); + /// The per-command bounds callers budget against (session-directory and + /// audio permit budgets in buzz-relay). Literal on purpose: raising a + /// timeout constant must fail these tests rather than move them with it. + const PROMISED_CONNECT_BOUND: Duration = Duration::from_millis(1000); + const PROMISED_RESPONSE_BOUND: Duration = Duration::from_millis(500); + + /// How a fake Redis peer treats one accepted socket. + #[derive(Clone, Copy)] + enum Peer { + /// Accept, then never answer anything (stalls the connect handshake). + Silent, + /// Answer the handshake, never answer a command. + WithholdReplies, + /// Answer the handshake and one PING, then close the socket. + OnePingThenClose, + /// Answer the handshake and one PING, then stay open and silent. + OnePingThenSilent, + /// Reject AUTH in the handshake. + RejectAuth, + /// Answer everything. + Healthy, + } + + /// Fake RESP peer on loopback. `script(n)` picks how the n-th accepted + /// socket (from 0) behaves. Returns its URL and an accept counter. + async fn fake_redis(script: fn(usize) -> Peer) -> (String, Arc) { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let url = format!("redis://{}", listener.local_addr().expect("addr")); + let accepted = Arc::new(AtomicUsize::new(0)); + let counter = Arc::clone(&accepted); + tokio::spawn(async move { + while let Ok((mut socket, _)) = listener.accept().await { + let peer = script(counter.fetch_add(1, Ordering::SeqCst)); + tokio::spawn(async move { + let mut buf = vec![0u8; 4096]; + let mut answered_ping = false; + loop { + let Ok(n) = socket.read(&mut buf).await else { + return; + }; + if n == 0 { + return; + } + if matches!(peer, Peer::Silent) { + continue; + } + let chunk = String::from_utf8_lossy(&buf[..n]).to_ascii_uppercase(); + // The handshake pipelines AUTH (with a password) and + // CLIENT SETINFO calls, in that order. + let auth = if matches!(peer, Peer::RejectAuth) { + "-WRONGPASS invalid username-password pair\r\n" + } else { + "+OK\r\n" + }; + let mut reply = auth.repeat(chunk.matches("AUTH").count()); + reply.push_str(&"+OK\r\n".repeat(chunk.matches("SETINFO").count())); + let pings = chunk.matches("PING").count(); + match peer { + Peer::Healthy => reply.push_str(&"+PONG\r\n".repeat(pings)), + Peer::OnePingThenClose if pings > 0 => { + reply.push_str("+PONG\r\n"); + let _ = socket.write_all(reply.as_bytes()).await; + return; // dropping the socket closes it + } + Peer::OnePingThenSilent if pings > 0 && !answered_ping => { + reply.push_str("+PONG\r\n"); + answered_ping = true; + } + _ => {} + } + if !reply.is_empty() && socket.write_all(reply.as_bytes()).await.is_err() { + return; + } + } + }); + } + }); + (url, accepted) + } + + async fn timed_ping(conn: &RedisConnection) -> (redis::RedisResult, Duration) { + let started = Instant::now(); + let result = redis::cmd("PING") + .query_async::(&mut conn.clone()) + .await; + (result, started.elapsed()) + } + + #[tokio::test] + async fn stalled_handshake_fails_each_command_at_the_connect_timeout() { + let (url, accepted) = fake_redis(|_| Peer::Silent).await; + let conn = redis_connection(&url).expect("lazy build never dials"); + for attempt in 1..=2 { + let (result, elapsed) = timed_ping(&conn).await; + assert!(result.is_err(), "attempt {attempt}: {result:?}"); + // The connect timeout fired (not an instant refusal), and no + // in-command retry stacked a second attempt on top of it. + assert!( + elapsed >= REDIS_CONNECT_TIMEOUT - Duration::from_millis(50), + "attempt {attempt} failed early: {elapsed:?}" + ); + assert!( + elapsed < PROMISED_CONNECT_BOUND + SLACK, + "attempt {attempt} waited {elapsed:?}" + ); + } + assert_eq!(accepted.load(Ordering::SeqCst), 2, "one dial per command"); + } + + #[tokio::test] + async fn withheld_reply_fails_the_command_at_the_response_timeout() { + let (url, _) = fake_redis(|_| Peer::WithholdReplies).await; + let conn = redis_connection(&url).expect("lazy build never dials"); + let (result, elapsed) = timed_ping(&conn).await; + let error = result.expect_err("no reply must fail the command"); + assert!(error.is_timeout(), "{error:?}"); + assert!( + elapsed >= REDIS_RESPONSE_TIMEOUT - Duration::from_millis(50), + "failed early: {elapsed:?}" + ); + assert!( + elapsed < PROMISED_RESPONSE_BOUND + SLACK, + "waited {elapsed:?}" + ); + } + + #[tokio::test] + async fn dropped_connection_is_redialed_by_a_later_command() { + let (url, accepted) = fake_redis(|_| Peer::OnePingThenClose).await; + let conn = redis_connection(&url).expect("lazy build never dials"); + let (first, _) = timed_ping(&conn).await; + assert_eq!(first.expect("first ping"), "PONG"); + // The peer closed the socket. Each later command is bounded, and the + // shared connection redials instead of staying dead. + let mut recovered = false; + for _ in 0..5 { + let (result, elapsed) = timed_ping(&conn).await; + assert!( + elapsed < PROMISED_CONNECT_BOUND + PROMISED_RESPONSE_BOUND + SLACK, + "waited {elapsed:?}" + ); + if result.is_ok() { + recovered = true; + break; + } + } + assert!(recovered, "no later command reconnected"); + assert!(accepted.load(Ordering::SeqCst) >= 2, "never redialed"); + } + + #[tokio::test] + async fn silent_socket_is_replaced_for_every_clone() { + // The first socket answers one PING, then stays open but never answers + // again (a dropped NAT entry, a failed-over endpoint). New sockets work. + let (url, accepted) = fake_redis(|n| { + if n == 0 { + Peer::OnePingThenSilent + } else { + Peer::Healthy + } + }) + .await; + let conn = redis_connection(&url).expect("lazy build never dials"); + let existing_clone = conn.clone(); + let (first, _) = timed_ping(&conn).await; + assert_eq!(first.expect("first ping"), "PONG"); + let (stalled, elapsed) = timed_ping(&conn).await; + let error = stalled.expect_err("the silent socket must time out"); + assert!(error.is_timeout(), "{error:?}"); + assert!( + elapsed < PROMISED_RESPONSE_BOUND + SLACK, + "waited {elapsed:?}" + ); + // The timeout retired the socket for every clone: the next command, + // through a handle cloned before the stall, dials once and succeeds. + let (recovered, _) = timed_ping(&existing_clone).await; + assert_eq!(recovered.expect("ping on a fresh socket"), "PONG"); + assert_eq!(accepted.load(Ordering::SeqCst), 2); + } + + #[tokio::test] + async fn failed_auth_handshake_is_retried_by_a_later_command() { + // The serving socket drops, the reconnect's AUTH is rejected (a + // credential rotation caught mid-way), then the server accepts again. + let (url, accepted) = fake_redis(|n| match n { + 0 => Peer::OnePingThenClose, + 1 => Peer::RejectAuth, + _ => Peer::Healthy, + }) + .await; + let url = url.replace("redis://", "redis://:pw@"); + let conn = redis_connection(&url).expect("lazy build never dials"); + let (first, _) = timed_ping(&conn).await; + assert_eq!(first.expect("first ping"), "PONG"); + let mut errors = Vec::new(); + let mut recovered = false; + for _ in 0..5 { + match timed_ping(&conn).await.0 { + Ok(pong) => { + assert_eq!(pong, "PONG"); + recovered = true; + break; + } + Err(error) => errors.push(error), + } + } + assert!( + errors + .iter() + .any(|e| e.kind() == redis::ErrorKind::AuthenticationFailed), + "the rejected handshake never surfaced: {errors:?}" + ); + // The AUTH failure was not cached: the same handle redialed past it. + assert!(recovered, "never recovered: {errors:?}"); + assert_eq!(accepted.load(Ordering::SeqCst), 3, "{errors:?}"); + } + + #[tokio::test] + async fn unreachable_redis_fails_each_command_within_the_bounds() { + // Port 1 refuses (or times out) on every CI runner; nothing listens. + let conn = redis_connection("redis://127.0.0.1:1").expect("lazy build never dials"); + for _ in 0..2 { + let (result, elapsed) = timed_ping(&conn).await; + assert!(result.is_err()); + assert!( + elapsed < PROMISED_CONNECT_BOUND + PROMISED_RESPONSE_BOUND + SLACK, + "command waited {elapsed:?}" + ); + } + } +} diff --git a/crates/buzz-pubsub/src/error.rs b/crates/buzz-pubsub/src/error.rs index 96bc1016c9..4346718f3a 100644 --- a/crates/buzz-pubsub/src/error.rs +++ b/crates/buzz-pubsub/src/error.rs @@ -7,10 +7,6 @@ pub enum PubSubError { #[error("Redis error: {0}")] Redis(#[from] redis::RedisError), - /// Failed to acquire a connection from the Redis pool. - #[error("Redis pool error: {0}")] - Pool(#[from] deadpool_redis::PoolError), - /// JSON serialization or deserialization failed. #[error("Serialization error: {0}")] Serialization(#[from] serde_json::Error), diff --git a/crates/buzz-pubsub/src/lib.rs b/crates/buzz-pubsub/src/lib.rs index 007a21ebef..57c6b90560 100644 --- a/crates/buzz-pubsub/src/lib.rs +++ b/crates/buzz-pubsub/src/lib.rs @@ -7,9 +7,9 @@ //! ```text //! buzz-relay process //! │ -//! ├── deadpool-redis pool → PUBLISH, SET, ZADD, etc. +//! ├── shared RedisConnection (one multiplexed socket) → PUBLISH, SET, ZADD, etc. //! │ -//! └── dedicated redis::aio::PubSub connection (NOT from pool) +//! └── dedicated redis::aio::PubSub connection (NOT the shared one) //! └── dynamic SUBSCRIBE buzz:{community}:channel:{id} / buzz:{community}:global //! └── run_subscriber() → broadcast::channel(4096) → N WS receivers //! ``` @@ -18,13 +18,18 @@ //! backoff (1s → 2s → 4s → … → 30s max). //! //! Dedicated pub/sub connection is stateful and cannot be shared. -//! Pool connections handle all other commands. +//! The shared [`RedisConnection`] handles all other commands. //! Lagged receivers get `RecvError::Lagged`. /// Cross-pod cache-key invalidation over Redis pub/sub. pub mod cache_invalidation; /// Cross-pod connection-control commands over Redis pub/sub. pub mod conn_control; +/// The shared Redis command connection. +pub mod connection; +pub use connection::{ + redis_connection, RedisConnection, REDIS_CONNECT_TIMEOUT, REDIS_RESPONSE_TIMEOUT, +}; /// Error types for pub/sub operations. pub mod error; /// Redis-backed NIP-98 replay seen-set. @@ -116,7 +121,7 @@ impl PubSubConfig { /// Central pub/sub manager for a Buzz relay instance. pub struct PubSubManager { - pool: deadpool_redis::Pool, + pool: crate::RedisConnection, /// Redis URL used by the reconnect loop to re-establish pub/sub connections. redis_url: String, /// Delay before unsubscribing after the last local interest is released. @@ -134,14 +139,14 @@ pub struct PubSubManager { impl PubSubManager { /// Creates a new `PubSubManager` connected to the given Redis URL. - pub async fn new(redis_url: &str, pool: deadpool_redis::Pool) -> Result { + pub async fn new(redis_url: &str, pool: crate::RedisConnection) -> Result { Self::with_config(PubSubConfig::new(redis_url), pool).await } /// Creates a new `PubSubManager` using explicit pub/sub configuration. pub async fn with_config( config: PubSubConfig, - pool: deadpool_redis::Pool, + pool: crate::RedisConnection, ) -> Result { let (broadcast_tx, _) = broadcast::channel(4096); let (cache_invalidation_tx, _) = broadcast::channel(4096); @@ -343,7 +348,7 @@ impl PubSubManager { ctx: &TenantContext, invalidation: &CacheInvalidation, ) -> Result { - let mut conn = self.pool.get().await?; + let mut conn = self.pool.clone(); let payload = serde_json::to_string(invalidation)?; let subscriber_count: i64 = redis::cmd("PUBLISH") .arg(cache_invalidation_channel(ctx)) @@ -363,7 +368,7 @@ impl PubSubManager { ctx: &TenantContext, command: &ConnControl, ) -> Result { - let mut conn = self.pool.get().await?; + let mut conn = self.pool.clone(); let payload = serde_json::to_string(command)?; let subscriber_count: i64 = redis::cmd("PUBLISH") .arg(conn_control_channel(ctx)) @@ -384,7 +389,7 @@ impl PubSubManager { command: &NipFiDisconnect, ) -> Result { use crate::conn_control::encode_nip_fi_disconnect; - let mut conn = self.pool.get().await?; + let mut conn = self.pool.clone(); let payload = encode_nip_fi_disconnect(command)?; let subscriber_count: i64 = redis::cmd("PUBLISH") .arg(channel) @@ -477,10 +482,8 @@ impl PubSubManager { #[cfg(test)] pub(crate) mod test_util { - pub fn make_test_pool() -> deadpool_redis::Pool { - let cfg = deadpool_redis::Config::from_url("redis://127.0.0.1:6379"); - cfg.create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("Failed to create Redis pool") + pub fn make_test_pool() -> crate::RedisConnection { + crate::redis_connection("redis://127.0.0.1:6379").expect("Failed to build Redis connection") } } @@ -599,7 +602,7 @@ mod tests { let status = presence::get_presence(&pool, &ctx, &pubkey).await.unwrap(); assert_eq!(status.as_deref(), Some("online")); - let mut conn = pool.get().await.unwrap(); + let mut conn = pool.clone(); let ttl: i64 = redis::cmd("TTL") .arg(presence::presence_key(&ctx, &pubkey)) .query_async(&mut conn) diff --git a/crates/buzz-pubsub/src/nip98_replay.rs b/crates/buzz-pubsub/src/nip98_replay.rs index 858f0f5ace..f59a07b56b 100644 --- a/crates/buzz-pubsub/src/nip98_replay.rs +++ b/crates/buzz-pubsub/src/nip98_replay.rs @@ -21,12 +21,12 @@ use nostr::EventId; /// the TTL window return `nil`, which we surface as `Ok(false)` so the caller /// rejects the request as replay. pub struct RedisNip98ReplayGuard { - pool: deadpool_redis::Pool, + pool: crate::RedisConnection, } impl RedisNip98ReplayGuard { - /// Create a new replay guard backed by the given Redis connection pool. - pub fn new(pool: deadpool_redis::Pool) -> Self { + /// Create a new replay guard backed by the shared Redis connection. + pub fn new(pool: crate::RedisConnection) -> Self { Self { pool } } } @@ -47,16 +47,7 @@ impl Nip98ReplayGuard for RedisNip98ReplayGuard { // implausibly long. let ttl = ttl_secs.clamp(DEFAULT_REPLAY_TTL_SECS, MAX_REPLAY_TTL_SECS); - let mut conn = self.pool.get().await.map_err(|e| { - // Structured field for ops; the user-facing AuthError stays a - // bounded category string. - tracing::warn!( - scope = %scope, - error = %e, - "nip98 replay: redis pool acquire failed — caller MUST fail closed" - ); - AuthError::Internal(format!("Redis pool: {e}")) - })?; + let mut conn = self.pool.clone(); let key = nip98_replay_key_for_scope(scope, event_id); @@ -69,7 +60,7 @@ impl Nip98ReplayGuard for RedisNip98ReplayGuard { .arg("NX") .arg("EX") .arg(ttl) - .query_async(&mut *conn) + .query_async(&mut conn) .await .map_err(|e| { tracing::warn!( @@ -102,15 +93,12 @@ impl Nip98ReplayGuard for RedisNip98ReplayGuard { mod tests { use super::*; use buzz_core::{CommunityId, TenantContext}; - use deadpool_redis::{Config, Runtime}; use nostr::{EventBuilder, Keys, Kind}; use uuid::Uuid; - fn redis_pool() -> deadpool_redis::Pool { + fn redis_pool() -> crate::RedisConnection { let url = std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379".into()); - Config::from_url(url) - .create_pool(Some(Runtime::Tokio1)) - .expect("create pool") + crate::redis_connection(&url).expect("redis connection") } fn fresh_ctx() -> TenantContext { diff --git a/crates/buzz-pubsub/src/nip_fi_command_replay.rs b/crates/buzz-pubsub/src/nip_fi_command_replay.rs index d21bd5c55d..47b3c0369d 100644 --- a/crates/buzz-pubsub/src/nip_fi_command_replay.rs +++ b/crates/buzz-pubsub/src/nip_fi_command_replay.rs @@ -15,13 +15,13 @@ use buzz_auth::{ /// `try_claim` issues `SET buzz:nip-fi:command:{hash} 1 NX EX `; `OK` /// is a first claim and `nil` is a replay. pub struct RedisCommandReplayGuard { - pool: deadpool_redis::Pool, + pool: crate::RedisConnection, prefix: &'static str, } impl RedisCommandReplayGuard { - /// Create a new replay guard backed by the given Redis connection pool. - pub fn new(pool: deadpool_redis::Pool) -> Self { + /// Create a new replay guard backed by the shared Redis connection. + pub fn new(pool: crate::RedisConnection) -> Self { Self { pool, prefix: COMMAND_REPLAY_PREFIX, @@ -30,19 +30,12 @@ impl RedisCommandReplayGuard { /// A guard for shadow pods: claims live under /// `buzz:nip-fi:shadow-command:{hash}`, disjoint from enforce claims. - pub fn shadow(pool: deadpool_redis::Pool) -> Self { + pub fn shadow(pool: crate::RedisConnection) -> Self { Self { pool, prefix: SHADOW_COMMAND_REPLAY_PREFIX, } } - - async fn conn(&self) -> Result { - self.pool.get().await.map_err(|e| { - tracing::warn!(error = %e, "nip-fi command replay: redis pool acquire failed"); - AuthError::Internal(format!("Redis pool: {e}")) - }) - } } impl CommandReplayGuard for RedisCommandReplayGuard { @@ -55,14 +48,14 @@ impl CommandReplayGuard for RedisCommandReplayGuard { { Box::pin(async move { let ttl = ttl_secs.clamp(DEFAULT_REPLAY_TTL_SECS, MAX_REPLAY_TTL_SECS); - let mut conn = self.conn().await?; + let mut conn = self.pool.clone(); let result: Option = redis::cmd("SET") .arg(namespaced_command_replay_key(self.prefix, issuer, jti)) .arg("1") .arg("NX") .arg("EX") .arg(ttl) - .query_async(&mut *conn) + .query_async(&mut conn) .await .map_err(|e| { tracing::warn!(error = %e, "nip-fi command replay: redis SET NX EX failed"); diff --git a/crates/buzz-pubsub/src/presence.rs b/crates/buzz-pubsub/src/presence.rs index 4de5fae8a3..67bc359832 100644 --- a/crates/buzz-pubsub/src/presence.rs +++ b/crates/buzz-pubsub/src/presence.rs @@ -4,8 +4,8 @@ //! TTL is 3x the 60s heartbeat interval so a single missed heartbeat doesn't //! cause presence flap. Clean disconnect deletes immediately. +use crate::RedisConnection; use buzz_core::TenantContext; -use deadpool_redis::Pool; use nostr::PublicKey; use std::collections::HashMap; @@ -26,12 +26,12 @@ pub fn presence_key(ctx: &TenantContext, pubkey: &PublicKey) -> String { /// Sets presence status for `pubkey` with a [`PRESENCE_TTL_SECS`]-second TTL. pub async fn set_presence( - pool: &Pool, + pool: &RedisConnection, ctx: &TenantContext, pubkey: &PublicKey, status: &str, ) -> Result<(), PubSubError> { - let mut conn = pool.get().await?; + let mut conn = pool.clone(); let key = presence_key(ctx, pubkey); redis::cmd("SET") .arg(&key) @@ -45,11 +45,11 @@ pub async fn set_presence( /// Removes the presence entry for `pubkey`. Call on clean disconnect. pub async fn clear_presence( - pool: &Pool, + pool: &RedisConnection, ctx: &TenantContext, pubkey: &PublicKey, ) -> Result<(), PubSubError> { - let mut conn = pool.get().await?; + let mut conn = pool.clone(); let key = presence_key(ctx, pubkey); redis::cmd("DEL") .arg(&key) @@ -61,7 +61,7 @@ pub async fn clear_presence( /// Sets one status for every pubkey in `pubkeys`, each with the /// [`PRESENCE_TTL_SECS`] TTL, in one atomic round trip. pub async fn set_presence_many( - pool: &Pool, + pool: &RedisConnection, ctx: &TenantContext, pubkeys: &[PublicKey], status: &str, @@ -69,7 +69,7 @@ pub async fn set_presence_many( if pubkeys.is_empty() { return Ok(()); } - let mut conn = pool.get().await?; + let mut conn = pool.clone(); let mut pipe = redis::pipe(); pipe.atomic(); for pubkey in pubkeys { @@ -86,14 +86,14 @@ pub async fn set_presence_many( /// Removes the presence entries for every pubkey in `pubkeys`. pub async fn clear_presence_many( - pool: &Pool, + pool: &RedisConnection, ctx: &TenantContext, pubkeys: &[PublicKey], ) -> Result<(), PubSubError> { if pubkeys.is_empty() { return Ok(()); } - let mut conn = pool.get().await?; + let mut conn = pool.clone(); let keys: Vec = pubkeys .iter() .map(|pubkey| presence_key(ctx, pubkey)) @@ -107,11 +107,11 @@ pub async fn clear_presence_many( /// Returns the current presence status for `pubkey`, or `None` if not set or expired. pub async fn get_presence( - pool: &Pool, + pool: &RedisConnection, ctx: &TenantContext, pubkey: &PublicKey, ) -> Result, PubSubError> { - let mut conn = pool.get().await?; + let mut conn = pool.clone(); let key = presence_key(ctx, pubkey); let value: Option = redis::cmd("GET").arg(&key).query_async(&mut conn).await?; Ok(value) @@ -119,14 +119,14 @@ pub async fn get_presence( /// Returns `pubkey_hex → status` for all currently-set keys. pub async fn get_presence_bulk( - pool: &Pool, + pool: &RedisConnection, ctx: &TenantContext, pubkeys: &[PublicKey], ) -> Result, PubSubError> { if pubkeys.is_empty() { return Ok(HashMap::new()); } - let mut conn = pool.get().await?; + let mut conn = pool.clone(); let keys: Vec = pubkeys .iter() .map(|pubkey| presence_key(ctx, pubkey)) @@ -161,10 +161,8 @@ mod tests { // A backend outage must surface as `Err`, not a silently-empty `Ok`. // `synthesize_presence` relies on this to return an error response // rather than a fake-empty "all offline" snapshot on a Redis failure. - // Pool points at a closed port so the connection attempt fails. - let pool = deadpool_redis::Config::from_url("redis://127.0.0.1:1") - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("pool builds lazily"); + // The connection points at a closed port so the connection attempt fails. + let pool = crate::redis_connection("redis://127.0.0.1:1").expect("pool builds lazily"); let ctx = ctx(0xaaaa, "a.example"); let pubkey = make_pubkey(); @@ -284,7 +282,7 @@ mod tests { result.get(&untouched.to_hex()).map(String::as_str), Some("away") ); - let mut conn = pool.get().await.unwrap(); + let mut conn = pool.clone(); let ttl: i64 = redis::cmd("TTL") .arg(presence_key(&ctx, &pk2)) .query_async(&mut conn) @@ -309,7 +307,7 @@ mod tests { set_presence(&pool, &ctx, &pubkey, "online").await.unwrap(); - let mut conn = pool.get().await.unwrap(); + let mut conn = pool.clone(); let ttl: i64 = redis::cmd("TTL") .arg(presence_key(&ctx, &pubkey)) .query_async(&mut conn) diff --git a/crates/buzz-pubsub/src/publisher.rs b/crates/buzz-pubsub/src/publisher.rs index 8ad06cc56b..85712f57b6 100644 --- a/crates/buzz-pubsub/src/publisher.rs +++ b/crates/buzz-pubsub/src/publisher.rs @@ -1,7 +1,7 @@ //! Event publishing — PUBLISH to Redis via pool connection. +use crate::RedisConnection; use buzz_core::TenantContext; -use deadpool_redis::Pool; use nostr::JsonUtil; use uuid::Uuid; @@ -20,12 +20,12 @@ pub fn global_key(ctx: &TenantContext) -> String { /// Returns the number of subscribers that received the message. pub async fn publish_event( - pool: &Pool, + pool: &RedisConnection, ctx: &TenantContext, topic: EventTopic, event: &nostr::Event, ) -> Result { - let mut conn = pool.get().await?; + let mut conn = pool.clone(); let key = crate::topic::EventTopicKey::from_context(ctx, topic).redis_channel(); let payload = event.as_json(); let subscriber_count: i64 = redis::cmd("PUBLISH") diff --git a/crates/buzz-pubsub/src/rate_limiter.rs b/crates/buzz-pubsub/src/rate_limiter.rs index 8d5494f1d5..095e6591e1 100644 --- a/crates/buzz-pubsub/src/rate_limiter.rs +++ b/crates/buzz-pubsub/src/rate_limiter.rs @@ -37,21 +37,18 @@ return {count, ttl} /// from a prior crash), the key is repaired with a fresh EXPIRE and a warning /// is logged. async fn run_rate_limit( - pool: &deadpool_redis::Pool, + pool: &crate::RedisConnection, key: &str, window_secs: u64, limit: u64, ) -> Result { - let mut conn = pool - .get() - .await - .map_err(|e| AuthError::Internal(format!("Redis pool: {e}")))?; + let mut conn = pool.clone(); let script = Script::new(RATE_LIMIT_SCRIPT); let (count, ttl): (u64, i64) = script .key(key) .arg(window_secs as i64) - .invoke_async(&mut *conn) + .invoke_async(&mut conn) .await .map_err(|e| AuthError::Internal(format!("Redis rate limit script: {e}")))?; @@ -62,7 +59,7 @@ async fn run_rate_limit( let _: () = redis::cmd("EXPIRE") .arg(key) .arg(window_secs as i64) - .query_async(&mut *conn) + .query_async(&mut conn) .await .map_err(|e| AuthError::Internal(format!("Redis EXPIRE repair: {e}")))?; // After repair, the window resets to the full duration. @@ -86,12 +83,12 @@ async fn run_rate_limit( /// managed atomically via a Lua script to prevent keys from persisting without /// expiry. pub struct RedisRateLimiter { - pool: deadpool_redis::Pool, + pool: crate::RedisConnection, } impl RedisRateLimiter { /// Create a new `RedisRateLimiter` backed by the given connection pool. - pub fn new(pool: deadpool_redis::Pool) -> Self { + pub fn new(pool: crate::RedisConnection) -> Self { Self { pool } } } diff --git a/crates/buzz-relay-mesh/Cargo.toml b/crates/buzz-relay-mesh/Cargo.toml index 5760f13770..72b8a7abb1 100644 --- a/crates/buzz-relay-mesh/Cargo.toml +++ b/crates/buzz-relay-mesh/Cargo.toml @@ -9,12 +9,12 @@ repository.workspace = true [dependencies] tokio = { workspace = true } +buzz-pubsub = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } postcard = { workspace = true } iroh = { workspace = true } redis = { workspace = true } -deadpool-redis = { workspace = true } thiserror = { workspace = true } tracing = { workspace = true } uuid = { workspace = true } diff --git a/crates/buzz-relay-mesh/src/registry.rs b/crates/buzz-relay-mesh/src/registry.rs index 635dff6328..4d8bd447cf 100644 --- a/crates/buzz-relay-mesh/src/registry.rs +++ b/crates/buzz-relay-mesh/src/registry.rs @@ -17,6 +17,11 @@ use sha2::{Digest, Sha256}; use crate::{MeshError, RuntimeId}; pub const READY_KEY_PREFIX: &str = "mesh:ready:"; +/// Sorted set of live ready-record keys, scored by expiry (unix seconds). +/// Reconcile reads this instead of SCANning the whole keyspace, so its cost +/// scales with the number of peers rather than with every key in Redis. The +/// name deliberately does not match `mesh:ready:*`. +pub const READY_INDEX_KEY: &str = "mesh:ready-index"; pub const DEFAULT_REGISTRY_REFRESH: Duration = Duration::from_secs(15); pub const REGISTRY_EXPIRY_MULTIPLIER: u64 = 3; pub const ATTESTATION_CONTEXT: &str = "buzz-relay-mesh-ready-v1"; @@ -155,16 +160,54 @@ pub fn expiry_for(refresh: Duration) -> Duration { refresh.saturating_mul(REGISTRY_EXPIRY_MULTIPLIER as u32) } +fn unix_now_secs() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0) +} + +/// Decode and authenticate the value stored at `key`; `None` (with a warn) +/// for anything a peer must not dial. +fn decode_ready_record(key: &str, raw: &str) -> Option { + match serde_json::from_str::(raw) { + Ok(record) if record.key() == key => match record.verify_attestation() { + Ok(()) => Some(record), + Err(err) => { + tracing::warn!( + key, + runtime_id = %record.runtime_id, + %err, + "mesh ready registry attestation failed — skipping" + ); + None + } + }, + Ok(record) => { + tracing::warn!( + key, + runtime_id = %record.runtime_id, + "mesh ready registry key/runtime mismatch — skipping" + ); + None + } + Err(err) => { + tracing::warn!(key, %err, "mesh ready registry decode failed — skipping"); + None + } + } +} + /// Redis-backed mesh bootstrap registry. #[derive(Clone)] pub struct ReadyRegistry { - pool: deadpool_redis::Pool, + redis: buzz_pubsub::RedisConnection, refresh: Duration, } impl ReadyRegistry { - pub fn new(pool: deadpool_redis::Pool, refresh: Duration) -> Self { - Self { pool, refresh } + pub fn new(redis: buzz_pubsub::RedisConnection, refresh: Duration) -> Self { + Self { redis, refresh } } pub fn refresh_interval(&self) -> Duration { @@ -181,15 +224,37 @@ impl ReadyRegistry { /// rule stays explicit at the relay boundary. pub async fn publish_ready(&self, record: &ReadyRecord) -> Result<(), MeshError> { record.verify_attestation()?; - let mut conn = self.conn().await?; + let mut conn = self.redis.clone(); let payload = serde_json::to_string(record) .map_err(|e| MeshError::Transport(format!("ready registry encode: {e}")))?; let ttl_secs = self.expiry().as_secs().max(1); - redis::cmd("SET") - .arg(record.key()) + let now = unix_now_secs(); + let key = record.key(); + // One MULTI: the record, its index entry, pruning of entries whose + // owners stopped refreshing, and an index TTL so a fully stopped fleet + // leaves nothing behind. + redis::pipe() + .atomic() + .cmd("SET") + .arg(&key) .arg(payload) .arg("EX") .arg(ttl_secs) + .ignore() + .cmd("ZADD") + .arg(READY_INDEX_KEY) + .arg(now + ttl_secs) + .arg(&key) + .ignore() + .cmd("ZREMRANGEBYSCORE") + .arg(READY_INDEX_KEY) + .arg("-inf") + .arg(format!("({now}")) + .ignore() + .cmd("EXPIRE") + .arg(READY_INDEX_KEY) + .arg(ttl_secs) + .ignore() .query_async::<()>(&mut conn) .await?; Ok(()) @@ -197,19 +262,56 @@ impl ReadyRegistry { /// Remove this runtime on clean shutdown. A crash is handled by TTL expiry. pub async fn clear_ready(&self, runtime_id: RuntimeId) -> Result<(), MeshError> { - let mut conn = self.conn().await?; - redis::cmd("DEL") - .arg(ready_key(runtime_id)) + let mut conn = self.redis.clone(); + let key = ready_key(runtime_id); + redis::pipe() + .atomic() + .cmd("DEL") + .arg(&key) + .ignore() + .cmd("ZREM") + .arg(READY_INDEX_KEY) + .arg(&key) + .ignore() .query_async::<()>(&mut conn) .await?; Ok(()) } - /// Scan all ready records. Malformed/stale/unauthenticated values are - /// skipped with a warn: a bad registry entry must not prevent bootstrap - /// from healthy peers. - pub async fn scan_ready(&self) -> Result, MeshError> { - let mut conn = self.conn().await?; + /// Read every indexed ready record: one ZRANGE plus one pipelined GET + /// round trip, independent of the size of the Redis keyspace. + /// Malformed/stale/unauthenticated values are skipped with a warn: a bad + /// registry entry must not prevent bootstrap from healthy peers. + pub async fn ready_records(&self) -> Result, MeshError> { + let mut conn = self.redis.clone(); + let keys: Vec = redis::cmd("ZRANGE") + .arg(READY_INDEX_KEY) + .arg(0) + .arg(-1) + .query_async(&mut conn) + .await?; + if keys.is_empty() { + return Ok(Vec::new()); + } + let mut pipe = redis::pipe(); + for key in &keys { + pipe.cmd("GET").arg(key); + } + let raws: Vec> = pipe.query_async(&mut conn).await?; + Ok(keys + .iter() + .zip(raws) + .filter_map(|(key, raw)| decode_ready_record(key, &raw?)) + .collect()) + } + + /// Full-keyspace SCAN for ready records, including those published by + /// relays that predate [`READY_INDEX_KEY`]. Cost is O(total Redis keys), + /// so the runtime runs it only at boot, on an unknown inbound peer, and + /// once every [`crate::runtime::LEGACY_SCAN_INTERVAL`]. Mixed-version compatibility only: delete it (and those call + /// sites) once every deployed relay writes the index. + pub async fn legacy_scan_ready(&self) -> Result, MeshError> { + let mut conn = self.redis.clone(); let mut cursor = 0u64; let mut out = Vec::new(); @@ -226,25 +328,8 @@ impl ReadyRegistry { for key in keys { let raw: Option = redis::cmd("GET").arg(&key).query_async(&mut conn).await?; - let Some(raw) = raw else { continue }; - match serde_json::from_str::(&raw) { - Ok(record) if record.key() == key => match record.verify_attestation() { - Ok(()) => out.push(record), - Err(err) => tracing::warn!( - key, - runtime_id = %record.runtime_id, - %err, - "mesh ready registry attestation failed — skipping" - ), - }, - Ok(record) => tracing::warn!( - key, - runtime_id = %record.runtime_id, - "mesh ready registry key/runtime mismatch — skipping" - ), - Err(err) => { - tracing::warn!(key, %err, "mesh ready registry decode failed — skipping") - } + if let Some(record) = raw.and_then(|raw| decode_ready_record(&key, &raw)) { + out.push(record); } } @@ -264,13 +349,6 @@ impl ReadyRegistry { published: false, } } - - async fn conn(&self) -> Result { - self.pool - .get() - .await - .map_err(|e| MeshError::Transport(format!("redis pool: {e}"))) - } } /// Readiness-gated registry heartbeat. @@ -324,6 +402,10 @@ mod tests { nostr::Keys::generate() } + fn lazy_redis(url: &str) -> buzz_pubsub::RedisConnection { + buzz_pubsub::redis_connection(url).expect("redis connection") + } + fn ready_record(byte: u8) -> ReadyRecord { ReadyRecord::new(rid(byte), &relay_keys(), vec![], 1, vec![]) } @@ -341,12 +423,12 @@ mod tests { assert_eq!(expiry_for(Duration::from_secs(15)), Duration::from_secs(45)); } - #[test] - fn heartbeat_starts_unpublished() { - let pool = deadpool_redis::Config::from_url("redis://127.0.0.1:6379") - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .unwrap(); - let registry = ReadyRegistry::new(pool, Duration::from_secs(15)); + #[tokio::test] + async fn heartbeat_starts_unpublished() { + let registry = ReadyRegistry::new( + lazy_redis("redis://127.0.0.1:6379"), + Duration::from_secs(15), + ); let heartbeat = registry.heartbeat(ready_record(1)); assert!(!heartbeat.published()); assert_eq!(heartbeat.record().runtime_id, rid(1)); @@ -382,4 +464,144 @@ mod tests { record.runtime_pubkey = rid(12).to_hex(); assert!(record.verify_attestation().is_err()); } + + /// Redis-backed registry tests. CI runs them in the relay integration lane + /// (`--run-ignored ignored-only`, `BUZZ_TEST_REDIS_URL`). They share the + /// fleet-wide index key, so they assert only on their own records. + mod external_infra_redis { + use super::*; + + fn pool() -> buzz_pubsub::RedisConnection { + let url = std::env::var("BUZZ_TEST_REDIS_URL") + .unwrap_or_else(|_| "redis://127.0.0.1:6379".to_string()); + lazy_redis(&url) + } + + fn fresh_record() -> ReadyRecord { + let runtime_id = RuntimeId(nostr::Keys::generate().public_key().to_bytes()); + ReadyRecord::new(runtime_id, &relay_keys(), vec![], 1, vec![]) + } + + fn ids(records: &[ReadyRecord]) -> Vec { + records.iter().map(|r| r.runtime_id).collect() + } + + async fn index_score(pool: &buzz_pubsub::RedisConnection, member: &str) -> Option { + let mut conn = pool.clone(); + redis::cmd("ZSCORE") + .arg(READY_INDEX_KEY) + .arg(member) + .query_async(&mut conn) + .await + .expect("zscore") + } + + #[tokio::test] + #[ignore = "requires Redis"] + async fn published_records_are_read_from_the_index_and_cleared_from_it() { + let pool = pool(); + let registry = ReadyRegistry::new(pool.clone(), Duration::from_secs(15)); + let (a, b) = (fresh_record(), fresh_record()); + registry.publish_ready(&a).await.unwrap(); + registry.publish_ready(&b).await.unwrap(); + + let indexed = ids(®istry.ready_records().await.unwrap()); + assert!(indexed.contains(&a.runtime_id) && indexed.contains(&b.runtime_id)); + let scanned = ids(®istry.legacy_scan_ready().await.unwrap()); + assert!(scanned.contains(&a.runtime_id) && scanned.contains(&b.runtime_id)); + + registry.clear_ready(a.runtime_id).await.unwrap(); + assert_eq!(index_score(&pool, &a.key()).await, None); + let indexed = ids(®istry.ready_records().await.unwrap()); + assert!(!indexed.contains(&a.runtime_id)); + assert!(indexed.contains(&b.runtime_id)); + assert!(!ids(®istry.legacy_scan_ready().await.unwrap()).contains(&a.runtime_id)); + + registry.clear_ready(b.runtime_id).await.unwrap(); + } + + #[tokio::test] + #[ignore = "requires Redis"] + async fn pre_index_records_are_found_only_by_the_legacy_scan() { + let pool = pool(); + let registry = ReadyRegistry::new(pool.clone(), Duration::from_secs(15)); + let legacy = fresh_record(); + { + // What a relay that predates the index writes. + let mut conn = pool.clone(); + redis::cmd("SET") + .arg(legacy.key()) + .arg(serde_json::to_string(&legacy).unwrap()) + .arg("EX") + .arg(45) + .query_async::<()>(&mut conn) + .await + .unwrap(); + } + + assert!(!ids(®istry.ready_records().await.unwrap()).contains(&legacy.runtime_id)); + assert!(ids(®istry.legacy_scan_ready().await.unwrap()).contains(&legacy.runtime_id)); + + registry.clear_ready(legacy.runtime_id).await.unwrap(); + } + + #[tokio::test] + #[ignore = "requires Redis"] + async fn publish_prunes_expired_index_entries_and_bounds_the_index() { + let pool = pool(); + let registry = ReadyRegistry::new(pool.clone(), Duration::from_secs(15)); + // An owner that stopped refreshing: its key has expired, its + // index entry's score is in the past. + let gone = ready_key(fresh_record().runtime_id); + { + let mut conn = pool.clone(); + redis::cmd("ZADD") + .arg(READY_INDEX_KEY) + .arg(unix_now_secs() - 1) + .arg(&gone) + .query_async::<()>(&mut conn) + .await + .unwrap(); + } + // Indexed but expired: skipped, not an error. + registry.ready_records().await.unwrap(); + + let fresh = fresh_record(); + registry.publish_ready(&fresh).await.unwrap(); + assert_eq!(index_score(&pool, &gone).await, None); + let score = index_score(&pool, &fresh.key()).await.expect("indexed"); + assert!(score > unix_now_secs() as f64); + + let mut conn = pool.clone(); + let ttl: i64 = redis::cmd("TTL") + .arg(READY_INDEX_KEY) + .query_async(&mut conn) + .await + .unwrap(); + assert!(ttl > 0 && ttl <= 45, "index TTL {ttl}"); + drop(conn); + + registry.clear_ready(fresh.runtime_id).await.unwrap(); + } + + #[tokio::test] + #[ignore = "requires Redis"] + async fn indexed_entry_whose_record_expired_is_skipped() { + let pool = pool(); + let registry = ReadyRegistry::new(pool.clone(), Duration::from_secs(15)); + let record = fresh_record(); + registry.publish_ready(&record).await.unwrap(); + { + let mut conn = pool.clone(); + redis::cmd("DEL") + .arg(record.key()) + .query_async::<()>(&mut conn) + .await + .unwrap(); + } + assert!(index_score(&pool, &record.key()).await.is_some()); + assert!(!ids(®istry.ready_records().await.unwrap()).contains(&record.runtime_id)); + registry.clear_ready(record.runtime_id).await.unwrap(); + } + } } diff --git a/crates/buzz-relay-mesh/src/runtime.rs b/crates/buzz-relay-mesh/src/runtime.rs index 091721c9a4..8347aec855 100644 --- a/crates/buzz-relay-mesh/src/runtime.rs +++ b/crates/buzz-relay-mesh/src/runtime.rs @@ -9,7 +9,7 @@ //! runtime id is present in the (attested) membership table; unknown ids get //! one registry rescan before rejection. Membership is a hint — admission //! here gates *dialability*, never session ownership. -//! - **Reconcile loop**: periodically rescans the Redis ready registry and +//! - **Reconcile loop**: periodically reads the Redis ready-registry index and //! dials every known, non-draining peer we are not yet connected to. This is //! what makes the mesh *warm*: failover is "next frame goes elsewhere," not //! "wait for a handshake." @@ -41,6 +41,13 @@ use crate::{InboundHandler, MeshDatagram, MeshError, MeshStream, RelayPeerTransp /// How often the reconcile loop rescans the registry and dials missing peers. pub const DEFAULT_RECONCILE_INTERVAL: Duration = Duration::from_secs(5); +/// How often a reconcile pass also runs the full-keyspace legacy SCAN, so peers +/// running a relay that predates the ready-record index stay in the table +/// during a mixed-version rollout (membership peer records expire after five +/// minutes). Measured in elapsed time, not passes: a pass waits on its dials, +/// so a pass count would stretch exactly while peers are unreachable. Delete +/// with [`ReadyRegistry::legacy_scan_ready`]. +pub const LEGACY_SCAN_INTERVAL: Duration = Duration::from_secs(60); /// How often each side sends a gossip digest on every control stream. pub const DEFAULT_GOSSIP_INTERVAL: Duration = Duration::from_secs(2); /// Bound on queued control-stream frames per peer before backpressure. @@ -154,7 +161,7 @@ impl MeshRuntime { /// Force one reconcile pass right now (bootstrap fast-path: dial the seed /// records without waiting for the first interval tick). pub async fn reconcile_now(&self) { - reconcile_once(&self.inner).await; + reconcile_once(&self.inner, RegistryRead::LegacyScan).await; } /// Stop all loops and drop all peer connections. @@ -321,29 +328,88 @@ async fn accept_loop(inner: Arc) { } /// Admission check for inbound connections: the runtime id must appear in the -/// attested membership table. Unknown ids get one registry rescan (covers the -/// bootstrap race where a fresh pod dials us before our next reconcile tick). +/// attested membership table. Unknown ids get one index read, then one legacy +/// scan (covers the bootstrap race where a fresh pod dials us before our next +/// reconcile tick, including a pod running a relay that predates the index). async fn is_known_peer(inner: &Arc, runtime_id: RuntimeId) -> bool { if inner.membership.has_peer(runtime_id) { return true; } - if let Some(registry) = &inner.registry { - match registry.scan_ready().await { - Ok(records) => inner.membership.apply_ready_records(records), - Err(err) => tracing::warn!(%err, "mesh: registry rescan on inbound failed"), + for read in [RegistryRead::Index, RegistryRead::LegacyScan] { + refresh_from_registry(inner, read).await; + if inner.membership.has_peer(runtime_id) { + return true; } } - inner.membership.has_peer(runtime_id) + false +} + +/// How a reconcile pass reads the ready registry. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum RegistryRead { + /// O(peers): the ready-record index. + Index, + /// O(total Redis keys): also finds peers that predate the index. + LegacyScan, +} + +impl RegistryRead { + /// Steady-state choice for a pass starting `since_last_scan` after the + /// previous legacy scan: the legacy scan once [`LEGACY_SCAN_INTERVAL`] has + /// elapsed, the index otherwise. + fn due(since_last_scan: Duration) -> Self { + if since_last_scan >= LEGACY_SCAN_INTERVAL { + Self::LegacyScan + } else { + Self::Index + } + } +} + +async fn refresh_from_registry(inner: &Arc, read: RegistryRead) { + let Some(registry) = &inner.registry else { + return; + }; + let result = match read { + RegistryRead::Index => registry.ready_records().await, + RegistryRead::LegacyScan => registry.legacy_scan_ready().await, + }; + match result { + Ok(records) => inner.membership.apply_ready_records(records), + Err(err) => tracing::warn!(%err, ?read, "mesh: registry read failed"), + } +} + +/// When the reconcile loop last ran the legacy scan. +struct LegacyScanClock { + last_scan: Instant, +} + +impl LegacyScanClock { + /// `reconcile_now` scans at boot, so the clock starts with the loop. + fn starting_at(now: Instant) -> Self { + Self { last_scan: now } + } + + /// The read for a pass starting at `now`; a due scan restarts the clock. + fn next_read(&mut self, now: Instant) -> RegistryRead { + let read = RegistryRead::due(now.saturating_duration_since(self.last_scan)); + if read == RegistryRead::LegacyScan { + self.last_scan = now; + } + read + } } async fn reconcile_loop(inner: Arc) { + let mut clock = LegacyScanClock::starting_at(Instant::now()); loop { - reconcile_once(&inner).await; + reconcile_once(&inner, clock.next_read(Instant::now())).await; tokio::time::sleep(inner.reconcile_interval).await; } } -async fn reconcile_once(inner: &Arc) { +async fn reconcile_once(inner: &Arc, read: RegistryRead) { let started = Instant::now(); { let peers = inner.peers.read().expect("peer lock poisoned"); @@ -356,12 +422,7 @@ async fn reconcile_once(inner: &Arc) { tracing::info!(pruned, "mesh: pruned expired peer records"); } - if let Some(registry) = &inner.registry { - match registry.scan_ready().await { - Ok(records) => inner.membership.apply_ready_records(records), - Err(err) => tracing::warn!(%err, "mesh: registry scan failed"), - } - } + refresh_from_registry(inner, read).await; inner.membership.publish_table_metrics(); let candidates = dial_candidates(inner); @@ -681,6 +742,45 @@ mod tests { SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0) } + #[test] + fn steady_state_reads_the_index_and_scans_on_elapsed_time() { + let cases = [ + (Duration::ZERO, RegistryRead::Index), + (DEFAULT_RECONCILE_INTERVAL, RegistryRead::Index), + ( + LEGACY_SCAN_INTERVAL - Duration::from_millis(1), + RegistryRead::Index, + ), + (LEGACY_SCAN_INTERVAL, RegistryRead::LegacyScan), + // One pass that sat on slow dials past the interval: the very next + // pass scans, however few passes have run. + (2 * LEGACY_SCAN_INTERVAL, RegistryRead::LegacyScan), + ]; + for (since_last_scan, want) in cases { + assert_eq!( + RegistryRead::due(since_last_scan), + want, + "{since_last_scan:?} since the last scan" + ); + } + // A legacy scan per interval keeps pre-index peers fresher than the + // membership peer-record TTL. + assert!(LEGACY_SCAN_INTERVAL < crate::membership::DEFAULT_PEER_RECORD_TTL); + } + + #[test] + fn a_legacy_scan_restarts_the_clock() { + let start = Instant::now(); + let mut clock = LegacyScanClock::starting_at(start); + let at = |secs| start + Duration::from_secs(secs); + assert_eq!(clock.next_read(at(5)), RegistryRead::Index); + assert_eq!(clock.next_read(at(60)), RegistryRead::LegacyScan); + // The next passes read the index until another interval elapses. + assert_eq!(clock.next_read(at(65)), RegistryRead::Index); + assert_eq!(clock.next_read(at(119)), RegistryRead::Index); + assert_eq!(clock.next_read(at(120)), RegistryRead::LegacyScan); + } + const RECONCILE_INTERVAL: Duration = Duration::from_millis(200); /// Long enough that the background reconcile loop never runs a second /// pass during a test, so the test drives every dial and state sync. diff --git a/crates/buzz-relay/Cargo.toml b/crates/buzz-relay/Cargo.toml index ef0808ef65..99350ab7ec 100644 --- a/crates/buzz-relay/Cargo.toml +++ b/crates/buzz-relay/Cargo.toml @@ -56,7 +56,6 @@ uuid = { workspace = true } chrono = { workspace = true } dashmap = { workspace = true } futures-util = { workspace = true } -deadpool-redis = { workspace = true } redis = { workspace = true } # Redis TLS (rediss://, e.g. ElastiCache) uses rustls, which needs a process # CryptoProvider installed at startup. Both aws-lc-rs and ring are compiled in diff --git a/crates/buzz-relay/src/api/admin/mod.rs b/crates/buzz-relay/src/api/admin/mod.rs index 85bdbf56d0..2199f8c9aa 100644 --- a/crates/buzz-relay/src/api/admin/mod.rs +++ b/crates/buzz-relay/src/api/admin/mod.rs @@ -1534,9 +1534,7 @@ mod postgres_tests { }); let pool = sqlx::PgPool::connect_lazy(&config.database_url).expect("lazy pg pool"); let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).expect("redis pool"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -2288,9 +2286,7 @@ mod postgres_tests { web_dir: None, }); let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).expect("redis pool"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -3096,9 +3092,7 @@ mod postgres_tests { }); let pool = sqlx::PgPool::connect_lazy(&config.database_url).expect("lazy pg pool"); let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).expect("redis pool"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -3590,9 +3584,7 @@ mod postgres_tests { }); let pool = sqlx::PgPool::connect_lazy(&config.database_url).expect("lazy pg pool"); let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).expect("redis pool"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -3668,9 +3660,7 @@ mod postgres_tests { .await .expect("connect to test DB"); let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).expect("redis pool"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -3795,9 +3785,7 @@ mod postgres_tests { drop(state); // not used let pool = sqlx::PgPool::connect_lazy(&config.database_url).expect("lazy pg pool"); let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).expect("redis pool"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -4264,9 +4252,7 @@ mod postgres_tests { }); let pool = sqlx::PgPool::connect_lazy(&config.database_url).expect("lazy pg pool"); let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).expect("redis pool"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -4337,9 +4323,7 @@ mod postgres_tests { }); let pool = sqlx::PgPool::connect_lazy(&config.database_url).expect("lazy pg pool"); let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).expect("redis pool"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -4407,9 +4391,7 @@ mod postgres_tests { }); let pool = sqlx::PgPool::connect_lazy(&config.database_url).expect("lazy pg pool"); let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).expect("redis pool"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -6209,9 +6191,7 @@ mod postgres_tests { web_dir: None, }); let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).expect("redis pool"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index a241e1cb58..f3898304ad 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -3441,11 +3441,9 @@ pub(crate) mod postgres_tests { use nostr::{Alphabet, EventBuilder, Keys, Kind, SingleLetterTag, Tag}; use std::sync::Mutex; - fn redis_pool() -> deadpool_redis::Pool { + fn redis_pool() -> buzz_pubsub::RedisConnection { let url = std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379".into()); - deadpool_redis::Config::from_url(url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("create redis pool") + buzz_pubsub::redis_connection(&url).expect("create redis pool") } fn fresh_tenant(host: &str) -> TenantContext { @@ -3540,9 +3538,8 @@ pub(crate) mod postgres_tests { use buzz_core::kind::KIND_PRESENCE_SNAPSHOT; // Pool at a closed port: get_presence_bulk's connection attempt fails. - let dead_pool = deadpool_redis::Config::from_url("redis://127.0.0.1:1") - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("pool builds lazily"); + let dead_pool = + buzz_pubsub::redis_connection("redis://127.0.0.1:1").expect("pool builds lazily"); let pubsub = buzz_pubsub::PubSubManager::new("redis://127.0.0.1:1", dead_pool) .await .expect("PubSubManager::new performs no IO"); @@ -4968,9 +4965,7 @@ pub(crate) mod postgres_tests { let pool = sqlx::PgPool::connect(database_url).await.ok()?; let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .ok()?; + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).ok()?; let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -5766,9 +5761,7 @@ pub(crate) mod postgres_tests { .await .ok()?; let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .ok()?; + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).ok()?; let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -5879,9 +5872,7 @@ pub(crate) mod postgres_tests { .await .ok()?; let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .ok()?; + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).ok()?; let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -5927,9 +5918,7 @@ pub(crate) mod postgres_tests { .await .ok()?; let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .ok()?; + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).ok()?; let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -6542,9 +6531,7 @@ pub(crate) mod postgres_tests { assert_one_auth_line(&lines, "denied", Some("quota_exceeded")); let mut unavailable = (*base).clone(); - let dead = deadpool_redis::Config::from_url("redis://127.0.0.1:1") - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let dead = buzz_pubsub::redis_connection("redis://127.0.0.1:1").expect("redis pool"); unavailable.admission_rate_limiter = Arc::new(buzz_pubsub::rate_limiter::RedisRateLimiter::new(dead)); let (status, lines) = @@ -9040,9 +9027,7 @@ pub(crate) mod postgres_tests { .await .ok()?; let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .ok()?; + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).ok()?; let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -9281,9 +9266,7 @@ pub(crate) mod postgres_tests { .await .ok()?; let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .ok()?; + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).ok()?; let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -9422,9 +9405,7 @@ pub(crate) mod postgres_tests { .await .ok()?; let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .ok()?; + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).ok()?; let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -10023,9 +10004,7 @@ pub(crate) mod postgres_tests { db.fence().force_open_for_tests(chrono::Utc::now()); db.set_replica_read_max_age_for_tests(Some(std::time::Duration::from_secs(5))); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).expect("redis pool"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -10261,9 +10240,7 @@ pub(crate) mod postgres_tests { let mut db = buzz_db::Db::from_pools(writer.clone(), replica.clone()); db.fence().force_open_for_tests(chrono::Utc::now()); db.set_replica_read_max_age_for_tests(Some(std::time::Duration::from_secs(5))); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).expect("redis pool"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await diff --git a/crates/buzz-relay/src/api/bridge/workflow_webhook_postgres_tests.rs b/crates/buzz-relay/src/api/bridge/workflow_webhook_postgres_tests.rs index fb5cd7ffc0..ad22536291 100644 --- a/crates/buzz-relay/src/api/bridge/workflow_webhook_postgres_tests.rs +++ b/crates/buzz-relay/src/api/bridge/workflow_webhook_postgres_tests.rs @@ -26,9 +26,7 @@ async fn app_state(db: buzz_db::Db, pool: &PgPool) -> (Arc, AuditShutd config.require_auth_token = false; config.require_relay_membership = false; - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("create Redis pool"); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).expect("create Redis pool"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await diff --git a/crates/buzz-relay/src/api/gifs.rs b/crates/buzz-relay/src/api/gifs.rs index 0dc247b2ee..f77b267807 100644 --- a/crates/buzz-relay/src/api/gifs.rs +++ b/crates/buzz-relay/src/api/gifs.rs @@ -368,9 +368,8 @@ mod tests { .connect_lazy("postgres://buzz:buzz_dev@127.0.0.1:1/buzz") // sadscan:disable np.postgres.1 .expect("lazy test database pool"); let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("lazy test Redis pool"); + let redis_pool = + buzz_pubsub::redis_connection(&config.redis_url).expect("lazy test Redis pool"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await diff --git a/crates/buzz-relay/src/api/git/policy.rs b/crates/buzz-relay/src/api/git/policy.rs index 1e376501ab..3716b4d883 100644 --- a/crates/buzz-relay/src/api/git/policy.rs +++ b/crates/buzz-relay/src/api/git/policy.rs @@ -835,9 +835,7 @@ printf '%s' "$HMAC_INPUT" | openssl dgst -sha256 -hmac "{secret}" -hex 2>/dev/nu .await .expect("connect test DB"); let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).expect("redis pool"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await diff --git a/crates/buzz-relay/src/api/git/settings_tests.rs b/crates/buzz-relay/src/api/git/settings_tests.rs index a1dd438e22..fd735520c0 100644 --- a/crates/buzz-relay/src/api/git/settings_tests.rs +++ b/crates/buzz-relay/src/api/git/settings_tests.rs @@ -67,9 +67,7 @@ mod postgres_tests { .await .ok()?; let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .ok()?; + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).ok()?; let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -243,9 +241,7 @@ mod postgres_tests { .await .ok()?; let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .ok()?; + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).ok()?; let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -344,9 +340,7 @@ mod postgres_tests { .await .ok()?; let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .ok()?; + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).ok()?; let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -1190,9 +1184,7 @@ mod external_infra { if std::env::var("BUZZ_TEST_SCHEMA_MODE").as_deref() != Ok("desired") { db.migrate().await.unwrap(); } - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .unwrap(); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).unwrap(); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await diff --git a/crates/buzz-relay/src/api/git/transport.rs b/crates/buzz-relay/src/api/git/transport.rs index 3fafe58eaf..35ae3cab4a 100644 --- a/crates/buzz-relay/src/api/git/transport.rs +++ b/crates/buzz-relay/src/api/git/transport.rs @@ -2572,9 +2572,7 @@ mod track_c_tests { .expect("connect test DB"); let db = buzz_db::Db::from_pool(pool.clone()); db.migrate().await.expect("migrate test DB"); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).expect("redis pool"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -4278,9 +4276,7 @@ mod off_mode_precedence_tests { let pool = sqlx::PgPool::connect(&config.database_url).await.ok()?; let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .ok()?; + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).ok()?; let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -4942,9 +4938,7 @@ mod off_mode_precedence_tests { Err(_) => panic!("local Postgres not reachable (enforce git)"), }; let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).expect("redis pool"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -5081,9 +5075,7 @@ mod off_mode_precedence_tests { Err(_) => panic!("local Postgres not reachable (pack routes)"), }; let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).expect("redis pool"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await diff --git a/crates/buzz-relay/src/api/invites.rs b/crates/buzz-relay/src/api/invites.rs index dc1709d32d..94650c7dbd 100644 --- a/crates/buzz-relay/src/api/invites.rs +++ b/crates/buzz-relay/src/api/invites.rs @@ -835,9 +835,7 @@ mod postgres_tests { let db = buzz_db::Db::from_pool(pool.clone()); db.ensure_configured_community(host).await.ok()?; - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .ok()?; + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).ok()?; let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await diff --git a/crates/buzz-relay/src/api/media.rs b/crates/buzz-relay/src/api/media.rs index 8b3095d5b5..0e9d5aaa3a 100644 --- a/crates/buzz-relay/src/api/media.rs +++ b/crates/buzz-relay/src/api/media.rs @@ -2068,9 +2068,7 @@ mod tests { .await .ok()?; let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .ok()?; + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).ok()?; let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -2155,9 +2153,7 @@ mod tests { .await .ok()?; let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .ok()?; + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).ok()?; let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -4484,9 +4480,7 @@ mod tests { db.ensure_configured_community("relay.example") .await .expect("seed relay.example community for host-bound media tests"); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).expect("redis pool"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await diff --git a/crates/buzz-relay/src/api/mesh_demo.rs b/crates/buzz-relay/src/api/mesh_demo.rs index 77dc9aaed8..e2b282b9b7 100644 --- a/crates/buzz-relay/src/api/mesh_demo.rs +++ b/crates/buzz-relay/src/api/mesh_demo.rs @@ -159,18 +159,16 @@ mod tests { use super::*; use crate::tunnel::directory::SessionDirectory; - fn pool() -> deadpool_redis::Pool { + fn pool() -> buzz_pubsub::RedisConnection { let url = std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379".into()); - deadpool_redis::Config::from_url(url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("create redis pool") + buzz_pubsub::redis_connection(&url).expect("create redis pool") } async fn redis_directory_if_available() -> Option { let pool = pool(); - let mut conn = pool.get().await.ok()?; + let mut conn = pool.clone(); redis::cmd("PING") - .query_async::(&mut *conn) + .query_async::(&mut conn) .await .ok()?; Some(SessionDirectory::with_lease_ttl( diff --git a/crates/buzz-relay/src/api/nip_fi.rs b/crates/buzz-relay/src/api/nip_fi.rs index e4e5f71cc3..fadcdbdbac 100644 --- a/crates/buzz-relay/src/api/nip_fi.rs +++ b/crates/buzz-relay/src/api/nip_fi.rs @@ -276,7 +276,7 @@ pub fn count_disconnect_event( /// Command replay guard for this pod. Shadow claims under its own key /// prefix, so a shadow accept never uses up the enforce claim. pub fn command_replay_guard( - pool: deadpool_redis::Pool, + pool: buzz_pubsub::RedisConnection, mode: NipFiMode, ) -> Arc { if mode.observes_only() { @@ -869,9 +869,7 @@ mod route_integration_tests { let pool = sqlx::PgPool::connect_lazy(&config.database_url).expect("lazy pg pool"); let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).expect("redis pool"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -971,9 +969,7 @@ mod route_integration_tests { let config = crate::config::Config::for_test(); let pool = sqlx::PgPool::connect_lazy(&config.database_url).unwrap(); let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .unwrap(); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).unwrap(); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -1232,9 +1228,7 @@ mod route_integration_tests { config.nip_fi.registry.insert(test_issuer_policy()); let pool = sqlx::PgPool::connect_lazy(&config.database_url).unwrap(); let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .unwrap(); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).unwrap(); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -1437,9 +1431,7 @@ mod route_integration_tests { config.nip_fi.registry.insert(test_issuer_policy()); let pool = sqlx::PgPool::connect_lazy(&config.database_url).unwrap(); let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .unwrap(); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).unwrap(); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -1632,9 +1624,7 @@ mod route_integration_tests { config.nip_fi.registry.insert(test_issuer_policy()); let pool = sqlx::PgPool::connect_lazy(&config.database_url).unwrap(); let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .unwrap(); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).unwrap(); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -1760,9 +1750,7 @@ mod route_integration_tests { let config = crate::config::Config::for_test(); let pool = sqlx::PgPool::connect_lazy(&config.database_url).unwrap(); let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .unwrap(); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).unwrap(); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -1951,9 +1939,7 @@ mod route_integration_tests { let pool = sqlx::PgPool::connect_lazy(&config.database_url).unwrap(); let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .unwrap(); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).unwrap(); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -2188,9 +2174,7 @@ mod route_integration_tests { config.nip_fi.registry.insert(test_issuer_policy()); let pool = sqlx::PgPool::connect_lazy(&config.database_url).unwrap(); let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .unwrap(); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).unwrap(); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -3626,9 +3610,7 @@ mod route_integration_tests { let url = std::env::var("BUZZ_TEST_REDIS_URL") .or_else(|_| std::env::var("REDIS_URL")) .unwrap_or_else(|_| "redis://127.0.0.1:6379".into()); - let pool = deadpool_redis::Config::from_url(url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let pool = buzz_pubsub::redis_connection(&url).expect("redis pool"); Arc::new(buzz_pubsub::RedisCommandReplayGuard::new(pool)) } @@ -3678,9 +3660,7 @@ mod route_integration_tests { }; let pool = sqlx::PgPool::connect_lazy(&config.database_url).unwrap(); let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .unwrap(); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).unwrap(); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await diff --git a/crates/buzz-relay/src/api/operator.rs b/crates/buzz-relay/src/api/operator.rs index 2558edd725..5854a0281a 100644 --- a/crates/buzz-relay/src/api/operator.rs +++ b/crates/buzz-relay/src/api/operator.rs @@ -1503,9 +1503,7 @@ mod postgres_tests { .ok()?; let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .ok()?; + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).ok()?; let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -4944,9 +4942,7 @@ mod postgres_tests { let pool = sqlx::PgPool::connect_lazy(&config.database_url).expect("lazy pg pool"); let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).expect("redis pool"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await diff --git a/crates/buzz-relay/src/audio/handler.rs b/crates/buzz-relay/src/audio/handler.rs index c3c2467c6c..b3b09cdb41 100644 --- a/crates/buzz-relay/src/audio/handler.rs +++ b/crates/buzz-relay/src/audio/handler.rs @@ -4339,9 +4339,7 @@ mod tests { .connect_lazy(&config.database_url) .expect("lazy pg pool"); let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).expect("redis pool"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -5929,9 +5927,7 @@ mod tests { configure(&mut config); let pool = sqlx::PgPool::connect_lazy(&config.database_url).expect("lazy pg pool"); let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).expect("redis pool"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -13686,9 +13682,7 @@ mod tests { let community = tenant.community(); let redis_url = std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379".into()); - let redis = deadpool_redis::Config::from_url(&redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let redis = buzz_pubsub::redis_connection(&redis_url).expect("redis pool"); let directory = SessionDirectory::with_db(redis, state.db.clone()); let (arrived_rx, release) = crate::nip_fi_test_hooks::directory_cas_hook::arm(community); @@ -13764,9 +13758,7 @@ mod tests { let relay_url = format!("ws://{}", tenant.host()); let redis_url = std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379".into()); - let redis = deadpool_redis::Config::from_url(&redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let redis = buzz_pubsub::redis_connection(&redis_url).expect("redis pool"); let directory = crate::tunnel::directory::SessionDirectory::with_db(redis, state.db.clone()); let mut mesh = @@ -13938,9 +13930,7 @@ mod tests { config.database_url = db_url.clone(); config.redis_url = "redis://127.0.0.1:1".to_string(); let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).expect("redis pool"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await diff --git a/crates/buzz-relay/src/audio/join.rs b/crates/buzz-relay/src/audio/join.rs index c200c20abc..7f43614943 100644 --- a/crates/buzz-relay/src/audio/join.rs +++ b/crates/buzz-relay/src/audio/join.rs @@ -402,15 +402,16 @@ const OWNER_READY_BUDGET: Duration = Duration::from_secs(2); /// Per-await bounds of the production [`SessionDirectory`] call chain. /// /// Each constant mirrors the awaits of one directory method, in order, and -/// each await's own bound: the whole-checkout Redis deadline, the redis 1.2 -/// driver's default 500 ms response timeout for one command (every script is a -/// single `EVAL`), and the serving-write lease SQL bound. +/// each await's own bound: the shared Redis connection's connect timeout (a +/// command issued while it is down waits for one attempt), its response +/// timeout for one command (every script is a single `EVAL`), and the +/// serving-write lease SQL bound. #[cfg(test)] pub(crate) mod worst_case { use std::time::Duration; - const CHECKOUT: Duration = crate::tunnel::directory::POOL_CHECKOUT_TIMEOUT; - const REDIS_COMMAND: Duration = Duration::from_millis(500); + const CHECKOUT: Duration = buzz_pubsub::REDIS_CONNECT_TIMEOUT; + const REDIS_COMMAND: Duration = buzz_pubsub::REDIS_RESPONSE_TIMEOUT; const LEASE_SQL: Duration = buzz_db::deletion::SERVING_WRITE_LEASE_SQL_TIMEOUT; const fn max(a: Duration, b: Duration) -> Duration { diff --git a/crates/buzz-relay/src/config.rs b/crates/buzz-relay/src/config.rs index 33074cc822..6aadf72d19 100644 --- a/crates/buzz-relay/src/config.rs +++ b/crates/buzz-relay/src/config.rs @@ -143,12 +143,6 @@ pub struct Config { pub drain_jitter_ms: u64, /// Redis connection URL used by the pub/sub manager. pub redis_url: String, - /// Maximum connections in the shared Redis pool. Defaults to 16. - /// - /// deadpool's own default is `CPU_COUNT * 2`, which on a 2-vCPU relay - /// pod is only 4 — small enough that rate-limit checks, presence, and - /// pub/sub publishes queue behind each other under load. - pub redis_pool_size: usize, /// Maximum connections in the Postgres writer/reader pools. Defaults to 50. /// /// The `buzz-db` default of 20 was sized for a handful of pods against @@ -731,12 +725,6 @@ impl Config { let redis_url = std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://localhost:6379".to_string()); - let redis_pool_size = std::env::var("BUZZ_REDIS_POOL_SIZE") - .ok() - .and_then(|v| v.parse::().ok()) - .filter(|&v| v > 0) - .unwrap_or(16); - let db_pool_size = std::env::var("BUZZ_DB_POOL_SIZE") .ok() .and_then(|v| v.parse::().ok()) @@ -1452,7 +1440,6 @@ impl Config { replica_read_max_age_ms, drain_jitter_ms, redis_url, - redis_pool_size, db_pool_size, db_read_pool_size, relay_url, @@ -1754,7 +1741,6 @@ mod tests { assert!(config.bind_addr.port() > 0); assert!(!config.database_url.is_empty()); assert!(!config.redis_url.is_empty()); - assert_eq!(config.redis_pool_size, 16); assert_eq!(config.db_pool_size, 50); assert!(config.max_connections > 0); assert!(config.send_buffer_size > 0); @@ -2351,31 +2337,6 @@ mod tests { )); } - #[test] - fn redis_pool_size_env_override_and_invalid_fallback() { - let _guards = env_guards(); - let previous = std::env::var_os("BUZZ_REDIS_POOL_SIZE"); - - std::env::set_var("BUZZ_REDIS_POOL_SIZE", "32"); - let overridden = Config::from_env().expect("config").redis_pool_size; - - std::env::set_var("BUZZ_REDIS_POOL_SIZE", "0"); - let zero = Config::from_env().expect("config").redis_pool_size; - - std::env::set_var("BUZZ_REDIS_POOL_SIZE", "not-a-number"); - let junk = Config::from_env().expect("config").redis_pool_size; - - if let Some(value) = previous { - std::env::set_var("BUZZ_REDIS_POOL_SIZE", value); - } else { - std::env::remove_var("BUZZ_REDIS_POOL_SIZE"); - } - - assert_eq!(overridden, 32); - assert_eq!(zero, 16, "zero must fall back to the default"); - assert_eq!(junk, 16, "unparsable value must fall back to the default"); - } - #[test] fn db_pool_size_env_override_and_invalid_fallback() { let _guards = env_guards(); diff --git a/crates/buzz-relay/src/handlers/auth.rs b/crates/buzz-relay/src/handlers/auth.rs index 76792fe649..1eb45dfb5d 100644 --- a/crates/buzz-relay/src/handlers/auth.rs +++ b/crates/buzz-relay/src/handlers/auth.rs @@ -1169,9 +1169,7 @@ mod tests { .connect_lazy(&config.database_url) .expect("lazy pg pool"); let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).expect("redis pool"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -1587,9 +1585,7 @@ mod tests { config.require_relay_membership = false; config.redis_url = "redis://127.0.0.1:1".to_string(); configure(&mut config); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).expect("redis pool"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -3195,9 +3191,7 @@ mod tests { config.database_url = db_url.clone(); config.redis_url = "redis://127.0.0.1:1".to_string(); let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).expect("redis pool"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -4116,9 +4110,7 @@ mod tests { config.database_url = db_url.clone(); config.redis_url = "redis://127.0.0.1:1".to_string(); let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).expect("redis pool"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await diff --git a/crates/buzz-relay/src/handlers/event.rs b/crates/buzz-relay/src/handlers/event.rs index 613b921c8f..9056dfe416 100644 --- a/crates/buzz-relay/src/handlers/event.rs +++ b/crates/buzz-relay/src/handlers/event.rs @@ -2677,10 +2677,8 @@ mod tests { async fn redis_url_if_available() -> Option { let redis_url = std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379".to_string()); - let pool = deadpool_redis::Config::from_url(&redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .ok()?; - let mut conn = pool.get().await.ok()?; + let pool = buzz_pubsub::redis_connection(&redis_url).ok()?; + let mut conn = pool.clone(); redis::cmd("PING") .query_async::(&mut conn) .await @@ -3031,9 +3029,7 @@ mod tests { config.redis_url = redis_url.to_string(); let pool = sqlx::PgPool::connect_lazy(&config.database_url).expect("lazy pg pool"); let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).expect("redis pool"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -3079,11 +3075,9 @@ mod tests { config.redis_url = "redis://127.0.0.1:6379".to_string(); let pool = sqlx::PgPool::connect(&config.database_url).await.ok()?; // Require a real Redis so dispatch's publish_event doesn't error-log. - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .ok()?; + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).ok()?; redis::cmd("PING") - .query_async::(&mut redis_pool.get().await.ok()?) + .query_async::(&mut redis_pool.clone()) .await .ok()?; let db = buzz_db::Db::from_pool(pool.clone()); diff --git a/crates/buzz-relay/src/handlers/identity_archive.rs b/crates/buzz-relay/src/handlers/identity_archive.rs index 919c1467b6..53f25cacee 100644 --- a/crates/buzz-relay/src/handlers/identity_archive.rs +++ b/crates/buzz-relay/src/handlers/identity_archive.rs @@ -444,9 +444,7 @@ mod tests { async fn test_state(pool: sqlx::PgPool) -> Option> { let db = buzz_db::Db::from_pool(pool.clone()); let config = crate::config::Config::for_test(); // [FI-TRACE-ENV-RACE] - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .ok()?; + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).ok()?; let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index eac849d48d..e0a8bd2289 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -6639,9 +6639,8 @@ mod postgres_tests { // ── AppState ───────────────────────────────────────────────────────── // Redis is lazy and never actually contacted on this rejection path. - let redis_pool = deadpool_redis::Config::from_url(FAKE_REDIS_URL) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("deadpool redis pool"); + let redis_pool = + buzz_pubsub::redis_connection(FAKE_REDIS_URL).expect("lazy redis connection"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(FAKE_REDIS_URL, redis_pool.clone()) .await @@ -6922,9 +6921,8 @@ mod postgres_tests { const FAKE_REDIS_URL: &str = "redis://127.0.0.1:1"; // never contacted let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(FAKE_REDIS_URL) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("deadpool redis pool"); + let redis_pool = + buzz_pubsub::redis_connection(FAKE_REDIS_URL).expect("lazy redis connection"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(FAKE_REDIS_URL, redis_pool.clone()) .await diff --git a/crates/buzz-relay/src/handlers/relay_admin.rs b/crates/buzz-relay/src/handlers/relay_admin.rs index bd5f7c56a4..1f51b16f89 100644 --- a/crates/buzz-relay/src/handlers/relay_admin.rs +++ b/crates/buzz-relay/src/handlers/relay_admin.rs @@ -771,9 +771,8 @@ mod postgres_tests { .expect("ensure community"); let tenant = TenantContext::resolved(record.id, host); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool config"); + let redis_pool = + buzz_pubsub::redis_connection(&config.redis_url).expect("redis pool config"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index d33d5fe10d..e6cd970a14 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -598,18 +598,14 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> { }; let step = StepTimer::start(StartupStep::RedisConnect); - let redis_pool = { - let mut cfg = deadpool_redis::Config::from_url(&config.redis_url); - cfg.pool = Some(deadpool_redis::PoolConfig::new(config.redis_pool_size)); - cfg.create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .map_err(|e| anyhow::anyhow!("Redis pool creation failed: {e}"))? - }; + // One-time bootstrap gate, deliberately before AppState and therefore before + // the health listener binds. Post-start Redis failures are dependency + // failures and must never move readiness; never having connected at all is + // a broken deployment, not a blip. + buzz_relay::state::verify_redis_command_path(&config.redis_url).await?; + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url) + .map_err(|e| anyhow::anyhow!("Redis connection setup failed: {e}"))?; let redis_health_pool = redis_pool.clone(); // cheap Arc clone — shared with AppState - // One-time bootstrap gate, deliberately before AppState and therefore before - // the health listener binds. Post-start Redis failures are dependency - // failures and must never move readiness; never having connected at all is - // a broken deployment, not a blip. - buzz_relay::state::verify_redis_command_path(&redis_health_pool).await?; info!("Redis command path connected"); let pubsub = Arc::new( PubSubManager::new(&config.redis_url, redis_pool) @@ -1438,7 +1434,8 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> { let router = build_router(Arc::clone(&state)); let health_router = build_health_router(Arc::clone(&state)); - // Pool metrics: periodic background task polling DB + Redis pool stats. + // Pool metrics: periodic background task polling DB pool stats and probing + // the shared Redis command connection. { let pool_state = Arc::clone(&state); let interval_secs = std::env::var("BUZZ_POOL_METRICS_INTERVAL_SECS") @@ -1484,11 +1481,7 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> { } } - let rs = pool_state.redis_pool.status(); - metrics::gauge!("buzz_redis_pool_available").set(rs.available as f64); - metrics::gauge!("buzz_redis_pool_size").set(rs.size as f64); - metrics::gauge!("buzz_redis_pool_max").set(rs.max_size as f64); - metrics::gauge!("buzz_redis_pool_waiting").set(rs.waiting as f64); + relay_metrics::probe_redis_command_connection(&pool_state.redis_pool).await; let deletion_store = pool_state.db.deletion_store(); match deletion_store.reap_expired_serving_write_leases(1000).await { diff --git a/crates/buzz-relay/src/mesh_boot.rs b/crates/buzz-relay/src/mesh_boot.rs index 5823bfdfb2..4c1eaea9ec 100644 --- a/crates/buzz-relay/src/mesh_boot.rs +++ b/crates/buzz-relay/src/mesh_boot.rs @@ -244,9 +244,8 @@ impl MeshHandle { } let transport: Arc = Arc::new(NoopTransport); - let pool = deadpool_redis::Config::from_url("redis://127.0.0.1:1") // never dialed - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("for_test_only: noop redis pool"); + let pool = buzz_pubsub::redis_connection("redis://127.0.0.1:1") // never dialed + .expect("for_test_only: lazy redis connection"); MeshHandle { directory: SessionDirectory::new(pool), @@ -533,7 +532,7 @@ fn advertise_addrs(endpoint: &MeshEndpoint) -> Vec { /// default tenant. pub async fn boot_mesh( config: &Config, - redis_pool: deadpool_redis::Pool, + redis_pool: buzz_pubsub::RedisConnection, db: buzz_db::Db, relay_keypair: &nostr::Keys, shutting_down: Arc, @@ -665,8 +664,7 @@ mod tests { async fn mesh_off_boots_nothing() { let mut config = crate::config::Config::for_test(); // [FI-TRACE-ENV-RACE] config.mesh.enabled = false; - let pool = deadpool_redis::Config::from_url("redis://127.0.0.1:1") // unroutable - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) + let pool = buzz_pubsub::redis_connection("redis://127.0.0.1:1") // unroutable .unwrap(); let keys = nostr::Keys::generate(); let db = buzz_db::Db::from_pool( @@ -705,8 +703,7 @@ mod tests { let mut config = crate::config::Config::for_test(); // [FI-TRACE-ENV-RACE] config.mesh.enabled = true; config.mesh.bind_addr = "127.0.0.1:0".parse().unwrap(); - let pool = deadpool_redis::Config::from_url("redis://127.0.0.1:1") // refused - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) + let pool = buzz_pubsub::redis_connection("redis://127.0.0.1:1") // refused .unwrap(); let db = buzz_db::Db::from_pool( sqlx::postgres::PgPoolOptions::new() @@ -814,9 +811,7 @@ mod tests { /// Returning to recv_frame after a tick loses that state and closes the stream. #[tokio::test(start_paused = true)] async fn demo_echo_retains_pending_receive_across_housekeeping_ticks() { - let pool = deadpool_redis::Config::from_url("redis://127.0.0.1:1") - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("test pool"); + let pool = buzz_pubsub::redis_connection("redis://127.0.0.1:1").expect("test pool"); let (started_tx, started_rx) = tokio::sync::oneshot::channel(); let (resume_tx, resume_rx) = tokio::sync::oneshot::channel(); let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0)); @@ -978,8 +973,7 @@ mod tests { let dispatcher = MeshInboundDispatcher::default(); let fence = Arc::new(crate::audio::mesh::GenerationFloor::new()); - let pool = deadpool_redis::Config::from_url("redis://127.0.0.1:1") // never dialed - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) + let pool = buzz_pubsub::redis_connection("redis://127.0.0.1:1") // never dialed .unwrap(); wire_mesh_consumers( &dispatcher, diff --git a/crates/buzz-relay/src/metrics.rs b/crates/buzz-relay/src/metrics.rs index a6ebbcd513..eebc313991 100644 --- a/crates/buzz-relay/src/metrics.rs +++ b/crates/buzz-relay/src/metrics.rs @@ -201,6 +201,11 @@ fn configured_prometheus_builder(gauge_idle_timeout_secs: u64) -> PrometheusBuil &READINESS_DURATION_BUCKETS_S, ) .expect("valid readiness duration bucket boundaries") + .set_buckets_for_metric( + Matcher::Full("buzz_redis_ping_seconds".to_owned()), + &READINESS_DURATION_BUCKETS_S, + ) + .expect("valid Redis PING duration bucket boundaries") .set_buckets_for_metric( Matcher::Full("buzz_db_pool_acquire_duration_seconds".to_owned()), &DB_POOL_ACQUIRE_DURATION_BUCKETS_S, @@ -313,6 +318,7 @@ pub fn try_install(port: u16, gauge_idle_timeout_secs: u64) -> Result<(), Metric describe_readiness_metrics(); describe_community_admission_metrics(); describe_db_pool_metrics(); + describe_redis_metrics(); describe_auth_metrics(); describe_ws_read_metrics(); crate::startup_steps::describe_metrics(); @@ -372,6 +378,48 @@ pub(crate) fn describe_community_admission_metrics() { ); } +/// Register the shared Redis command-connection probe. +pub(crate) fn describe_redis_metrics() { + metrics::describe_histogram!( + "buzz_redis_ping_seconds", + metrics::Unit::Seconds, + "Round trip of a periodic PING on the shared Redis command connection, including time queued behind other commands" + ); + metrics::describe_counter!( + "buzz_redis_ping_failures_total", + "Periodic PINGs on the shared Redis command connection that failed or timed out" + ); +} + +/// PING the shared Redis command connection once and record the round trip. +/// +/// Every Redis command shares that connection, so the round trip is the +/// delay an ordinary command sees right now. Bounded by the connection's +/// connect and response timeouts. +pub async fn probe_redis_command_connection(connection: &buzz_pubsub::RedisConnection) { + let started = std::time::Instant::now(); + let ping = redis::cmd("PING") + .query_async::(&mut connection.clone()) + .await; + match ping { + Ok(_) => record_redis_ping(Some(started.elapsed())), + Err(error) => { + tracing::warn!(%error, "Redis PING probe failed"); + record_redis_ping(None); + } + } +} + +/// Record one probe result: the round trip, or `None` for a failure. +fn record_redis_ping(round_trip: Option) { + match round_trip { + Some(elapsed) => { + metrics::histogram!("buzz_redis_ping_seconds").record(elapsed.as_secs_f64()) + } + None => metrics::counter!("buzz_redis_ping_failures_total").increment(1), + } +} + /// Register the frozen operation-aware pool-acquisition contract. pub(crate) fn describe_db_pool_metrics() { metrics::describe_counter!( @@ -720,6 +768,47 @@ mod contract_tests { const OUTCOMES: [&str; 4] = ["success", "timeout", "error", "cancelled"]; + #[test] + fn redis_ping_probe_exports_bucketed_round_trips_and_failures() { + let (recorder, handle) = super::readiness_test_recorder(); + metrics::with_local_recorder(&recorder, || { + super::describe_redis_metrics(); + super::record_redis_ping(Some(Duration::from_millis(3))); + super::record_redis_ping(None); + }); + let scrape = handle.render(); + assert!( + scrape.contains("buzz_redis_ping_seconds_bucket{le=\"0.005\"} 1"), + "{scrape}" + ); + assert!( + scrape.contains("buzz_redis_ping_seconds_count 1"), + "{scrape}" + ); + assert!( + scrape.contains("buzz_redis_ping_failures_total 1"), + "{scrape}" + ); + } + + #[tokio::test] + async fn redis_ping_probe_counts_an_unreachable_connection_as_a_failure() { + let (recorder, handle) = super::readiness_test_recorder(); + let connection = + buzz_pubsub::redis_connection("redis://127.0.0.1:1").expect("lazy connection"); + let _guard = metrics::set_default_local_recorder(&recorder); + super::probe_redis_command_connection(&connection).await; + let scrape = handle.render(); + assert!( + scrape.contains("buzz_redis_ping_failures_total 1"), + "{scrape}" + ); + assert!( + !scrape.contains("buzz_redis_ping_seconds_count"), + "{scrape}" + ); + } + fn label_keys(line: &str) -> BTreeSet<&str> { line.split_once('{') .and_then(|(_, rest)| rest.split_once('}')) diff --git a/crates/buzz-relay/src/readiness.rs b/crates/buzz-relay/src/readiness.rs index f3afffde79..94bed5099c 100644 --- a/crates/buzz-relay/src/readiness.rs +++ b/crates/buzz-relay/src/readiness.rs @@ -367,14 +367,23 @@ where DependencyReport::for_dependencies(postgres, redis, deletion_catalog, started_at.elapsed()) } -async fn redis_check(pool: &deadpool_redis::Pool, deadline: Instant) -> RedisOutcome { - match tokio::time::timeout_at(deadline, pool.get()).await { +/// PING the shared command connection. The `pool_*` outcome labels predate +/// the shared connection and are kept so existing dashboards keep matching. +async fn redis_check(pool: &buzz_pubsub::RedisConnection, deadline: Instant) -> RedisOutcome { + let mut connection = pool.clone(); + let ping = async { + redis::cmd("PING") + .query_async::(&mut connection) + .await + }; + match tokio::time::timeout_at(deadline, ping).await { Err(_) => RedisOutcome::PoolTimeout, + Ok(Err(error)) if error.is_timeout() => RedisOutcome::PoolTimeout, Ok(Err(error)) => { - tracing::debug!(error = %error, "Redis readiness pool acquisition failed"); + tracing::debug!(error = %error, "Redis readiness PING failed"); RedisOutcome::PoolError } - Ok(Ok(_connection)) => RedisOutcome::Success, + Ok(Ok(_)) => RedisOutcome::Success, } } @@ -398,14 +407,22 @@ fn classify_deletion_catalog_result(result: buzz_db::Result<()>) -> DeletionCata #[async_trait::async_trait] pub(crate) trait DependencyEvaluator: Send + Sync { - async fn evaluate(&self, db: &Db, redis_pool: &deadpool_redis::Pool) -> DependencyReport; + async fn evaluate( + &self, + db: &Db, + redis_pool: &buzz_pubsub::RedisConnection, + ) -> DependencyReport; } struct ProductionDependencyEvaluator; #[async_trait::async_trait] impl DependencyEvaluator for ProductionDependencyEvaluator { - async fn evaluate(&self, db: &Db, redis_pool: &deadpool_redis::Pool) -> DependencyReport { + async fn evaluate( + &self, + db: &Db, + redis_pool: &buzz_pubsub::RedisConnection, + ) -> DependencyReport { let deadline = Instant::now() + DEPENDENCY_TIMEOUT; evaluate_dependencies( async { db.readiness_check(deadline).await.into() }, @@ -491,7 +508,7 @@ impl DependencyDiagnostics { /// The completion timestamp is written last, once the cache already serves /// this report, so the gauge can never describe a sample `/_status` is not /// yet answering with. - pub(crate) async fn sample(&self, db: &Db, redis_pool: &deadpool_redis::Pool) { + pub(crate) async fn sample(&self, db: &Db, redis_pool: &buzz_pubsub::RedisConnection) { let report = self.evaluator.evaluate(db, redis_pool).await; record_dependency_report(&report); let sample = DependencySample { @@ -595,7 +612,7 @@ pub fn start_dependency_sampler_and_completion_publisher( async fn run_dependency_sampler_for_diagnostics( diagnostics: Arc, db: Db, - redis_pool: deadpool_redis::Pool, + redis_pool: buzz_pubsub::RedisConnection, sample_interval: Duration, cancel: CancellationToken, ) { @@ -944,11 +961,9 @@ mod tests { /// A `Db` and a Redis pool on a closed port. The scripted evaluators below /// never touch either, so no connection is ever attempted; they exist only /// to satisfy the production `sample` signature. - fn unreachable_dependencies() -> (Db, deadpool_redis::Pool) { + fn unreachable_dependencies() -> (Db, buzz_pubsub::RedisConnection) { let pool = sqlx::PgPool::connect_lazy("postgres://127.0.0.1:1/buzz").expect("lazy pg pool"); - let redis_pool = deadpool_redis::Config::from_url("redis://127.0.0.1:1") - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let redis_pool = buzz_pubsub::redis_connection("redis://127.0.0.1:1").expect("redis pool"); (Db::from_pool(pool), redis_pool) } @@ -956,7 +971,11 @@ mod tests { #[async_trait::async_trait] impl DependencyEvaluator for FixedEvaluator { - async fn evaluate(&self, _db: &Db, _redis_pool: &deadpool_redis::Pool) -> DependencyReport { + async fn evaluate( + &self, + _db: &Db, + _redis_pool: &buzz_pubsub::RedisConnection, + ) -> DependencyReport { self.0 } } @@ -969,7 +988,11 @@ mod tests { #[async_trait::async_trait] impl DependencyEvaluator for DelayedEvaluator { - async fn evaluate(&self, _db: &Db, _redis_pool: &deadpool_redis::Pool) -> DependencyReport { + async fn evaluate( + &self, + _db: &Db, + _redis_pool: &buzz_pubsub::RedisConnection, + ) -> DependencyReport { self.started.notify_one(); self.release.notified().await; self.report diff --git a/crates/buzz-relay/src/rejection/quota_tests.rs b/crates/buzz-relay/src/rejection/quota_tests.rs index ac66b11638..a606161d1d 100644 --- a/crates/buzz-relay/src/rejection/quota_tests.rs +++ b/crates/buzz-relay/src/rejection/quota_tests.rs @@ -24,9 +24,7 @@ async fn state_with_limits(messages: u64, agent_messages: u64, ws: u64) -> AppSt ..RateLimitConfig::default() }; state.auth = Arc::new(AuthService::new(auth)); - state.redis_pool = deadpool_redis::Config::from_url(url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .unwrap(); + state.redis_pool = buzz_pubsub::redis_connection(&url).unwrap(); state.admission_rate_limiter = Arc::new(RedisRateLimiter::new(state.redis_pool.clone())); state } @@ -57,10 +55,10 @@ fn pubkey(conn: &ConnectionState) -> nostr::PublicKey { async fn counter(state: &AppState, conn: &ConnectionState, bucket: LimitType) -> Option { let key = buzz_auth::rate_limit::rate_limit_key(&conn.tenant, &pubkey(conn), &bucket); - let mut redis = state.redis_pool.get().await.unwrap(); + let mut redis = state.redis_pool.clone(); redis::cmd("GET") .arg(key) - .query_async(&mut *redis) + .query_async(&mut redis) .await .unwrap() } diff --git a/crates/buzz-relay/src/router.rs b/crates/buzz-relay/src/router.rs index a7861a9b78..278ca65179 100644 --- a/crates/buzz-relay/src/router.rs +++ b/crates/buzz-relay/src/router.rs @@ -1322,7 +1322,7 @@ mod tests { async fn evaluate( &self, _db: &buzz_db::Db, - _redis_pool: &deadpool_redis::Pool, + _redis_pool: &buzz_pubsub::RedisConnection, ) -> DependencyReport { self.evaluations_started.fetch_add(1, Ordering::SeqCst); self.evaluations @@ -1398,9 +1398,7 @@ mod tests { }); let pool = sqlx::PgPool::connect_lazy(&config.database_url).expect("lazy pg pool"); let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).expect("redis pool"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -1446,9 +1444,7 @@ mod tests { config.redis_url = "redis://127.0.0.1:1".to_string(); let pool = sqlx::PgPool::connect_lazy(&config.database_url).expect("lazy pg pool"); let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).expect("redis pool"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -1708,7 +1704,7 @@ mod tests { async fn evaluate( &self, _db: &buzz_db::Db, - _redis_pool: &deadpool_redis::Pool, + _redis_pool: &buzz_pubsub::RedisConnection, ) -> DependencyReport { self.started.fetch_add(1, Ordering::SeqCst); let in_flight = self.in_flight.fetch_add(1, Ordering::SeqCst) + 1; @@ -2432,9 +2428,7 @@ mod tests { .connect_lazy(&config.database_url) .expect("lazy pg pool"); let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).expect("redis pool"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -3033,10 +3027,9 @@ mod tests { ); // Real Redis so admission reaches the handler, not the // fail-closed rate limiter. - let redis = deadpool_redis::Config::from_url( - std::env::var("REDIS_URL").unwrap_or("redis://127.0.0.1:6379".into()), + let redis = buzz_pubsub::redis_connection( + &std::env::var("REDIS_URL").unwrap_or("redis://127.0.0.1:6379".into()), ) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) .unwrap(); state.admission_rate_limiter = Arc::new( buzz_pubsub::rate_limiter::RedisRateLimiter::new(redis.clone()), @@ -3209,9 +3202,7 @@ mod tests { config.redis_url = "redis://127.0.0.1:1".to_string(); config.database_url = db_url; let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).expect("redis pool"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -3794,9 +3785,7 @@ mod tests { let pool = sqlx::PgPool::connect_lazy(&config.database_url).expect("lazy pg pool"); let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).expect("redis pool"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -3895,9 +3884,7 @@ mod tests { let pool = sqlx::PgPool::connect_lazy(&config.database_url).expect("lazy pg pool"); let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).expect("redis pool"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await @@ -4299,9 +4286,7 @@ mod tests { .connect_lazy(&config.database_url) .expect("lazy pg pool"); let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).expect("redis pool"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index dda0448170..28f107b3b1 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -28,7 +28,6 @@ use buzz_pubsub::{PubSubManager, RedisNip98ReplayGuard}; use buzz_search::SearchService; use buzz_workflow::WorkflowEngine; use chrono::{DateTime, Utc}; -use deadpool_redis; use crate::audio::AudioRoomManager; use crate::config::Config; @@ -603,24 +602,28 @@ const REDIS_BOOTSTRAP_DRIVER_TIMEOUT: std::time::Duration = /// Proves once, during startup, that the Redis command path this pod will serve /// from can actually be reached. /// -/// `deadpool_redis` pools dial lazily and `PubSubManager::new` only allocates -/// channels, so without this nothing in boot ever opened a command connection: -/// a relay came up against a dead Redis, bound its health listener, and — since -/// readiness reports local lifecycle only — advertised ready forever. Binding -/// that listener is a one-way latch, so the check has to happen before it, and -/// it is deliberately a *startup* gate: once serving, a Redis blip is a -/// dependency failure and must never change readiness. -pub async fn verify_redis_command_path(pool: &deadpool_redis::Pool) -> anyhow::Result<()> { +/// The shared command connection dials lazily and `PubSubManager::new` only +/// allocates channels, so without this nothing in boot ever opened a command +/// connection: a relay came up against a dead Redis, bound its health +/// listener, and — since readiness reports local lifecycle only — advertised +/// ready forever. Binding that listener is a one-way latch, so the check has to +/// happen before it, and it is deliberately a *startup* gate: once serving, a +/// Redis blip is a dependency failure and must never change readiness. +/// +/// The PING runs on a one-shot connection to the same URL so its longer driver +/// timeouts never apply to serving traffic; the outer timeout bounds connect, +/// handshake, and PING together. +pub async fn verify_redis_command_path(redis_url: &str) -> anyhow::Result<()> { let ping = async { - let connection = pool - .get() + let client = redis::Client::open(redis_url) + .map_err(|error| anyhow::anyhow!("{REDIS_BOOTSTRAP_FAILURE}: {error}"))?; + let config = redis::AsyncConnectionConfig::new() + .set_connection_timeout(Some(REDIS_BOOTSTRAP_DRIVER_TIMEOUT)) + .set_response_timeout(Some(REDIS_BOOTSTRAP_DRIVER_TIMEOUT)); + let mut connection = client + .get_multiplexed_async_connection_with_config(&config) .await .map_err(|error| anyhow::anyhow!("{REDIS_BOOTSTRAP_FAILURE}: {error}"))?; - // This one-shot connection is removed from the pool so extending its - // driver timeout cannot leak into normal serving traffic. The relay's - // outer timeout below must bound both lazy checkout and PING. - let mut connection = deadpool_redis::Connection::take(connection); - connection.set_response_timeout(REDIS_BOOTSTRAP_DRIVER_TIMEOUT); redis::cmd("PING") .query_async::(&mut connection) .await @@ -1394,7 +1397,7 @@ pub struct AppState { /// Database connection pool. pub db: Db, /// Redis pool for readiness health checks. - pub redis_pool: deadpool_redis::Pool, + pub redis_pool: buzz_pubsub::RedisConnection, /// Audit event service, absent when audit logging is disabled. pub audit: Option>, /// Pub/sub manager for broadcasting events to subscribers. @@ -1682,7 +1685,7 @@ impl AppState { pub fn new( config: Config, db: Db, - redis_pool: deadpool_redis::Pool, + redis_pool: buzz_pubsub::RedisConnection, audit: impl Into>, pubsub: Arc, auth: AuthService, @@ -3541,9 +3544,7 @@ pub(crate) mod tests { async fn build_test_state(config: crate::config::Config, pool: sqlx::PgPool) -> Arc { let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).expect("redis pool"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await diff --git a/crates/buzz-relay/src/tunnel/directory.rs b/crates/buzz-relay/src/tunnel/directory.rs index 5a60a0f984..86cba78c47 100644 --- a/crates/buzz-relay/src/tunnel/directory.rs +++ b/crates/buzz-relay/src/tunnel/directory.rs @@ -12,8 +12,6 @@ use buzz_relay_mesh::{FencedHeader, MeshError, Profile, RuntimeId}; use uuid::Uuid; const DEFAULT_LEASE_TTL: Duration = Duration::from_secs(30); -/// Bound on one Redis pool checkout (wait, create, or recycle). -pub(crate) const POOL_CHECKOUT_TIMEOUT: Duration = Duration::from_secs(1); const ACQUIRE_SCRIPT: &str = r#" local lease_key = KEYS[1] @@ -100,7 +98,7 @@ return {current, known_generation} /// Redis-backed owner directory for mesh tunnel sessions. #[derive(Clone)] pub struct SessionDirectory { - pool: deadpool_redis::Pool, + pool: buzz_pubsub::RedisConnection, lease_ttl: Duration, db: Option, } @@ -160,9 +158,6 @@ pub enum ReleaseResult { /// Errors from the Redis session directory. #[derive(Debug, thiserror::Error)] pub enum DirectoryError { - /// Redis pool checkout failed. - #[error("redis pool: {0}")] - Pool(#[from] deadpool_redis::PoolError), /// Redis command/script failed. #[error("redis: {0}")] Redis(#[from] redis::RedisError), @@ -202,13 +197,13 @@ pub enum DirectoryError { impl SessionDirectory { /// Create a directory backed by `pool` with the default lease TTL. - pub fn new(pool: deadpool_redis::Pool) -> Self { + pub fn new(pool: buzz_pubsub::RedisConnection) -> Self { Self::with_lease_ttl(pool, DEFAULT_LEASE_TTL) } /// Create a serving directory whose Redis mutations use durable, /// heartbeat-backed community write leases. - pub fn with_db(pool: deadpool_redis::Pool, db: buzz_db::Db) -> Self { + pub fn with_db(pool: buzz_pubsub::RedisConnection, db: buzz_db::Db) -> Self { Self { pool, lease_ttl: DEFAULT_LEASE_TTL, @@ -217,7 +212,7 @@ impl SessionDirectory { } /// Create a directory backed by `pool` with an explicit lease TTL. - pub fn with_lease_ttl(pool: deadpool_redis::Pool, lease_ttl: Duration) -> Self { + pub fn with_lease_ttl(pool: buzz_pubsub::RedisConnection, lease_ttl: Duration) -> Self { Self { pool, lease_ttl, @@ -225,22 +220,6 @@ impl SessionDirectory { } } - /// Check out a Redis connection within [`POOL_CHECKOUT_TIMEOUT`] in total. - /// - /// Callers may hold a session effect permit, so an exhausted or unhealthy - /// pool must end as an ordinary error rather than an indefinite wait. One - /// outer deadline covers the permit wait, creation, and deadpool's - /// retry-on-failed-recycle loop, which per-stage `Timeouts` would each reset. - /// Dropping `get()` is safe: deadpool 0.13 returns the permit and detaches - /// any half-recycled or half-created object. - async fn connection(&self) -> Result { - tokio::time::timeout(POOL_CHECKOUT_TIMEOUT, self.pool.get()) - .await - .unwrap_or(Err(deadpool_redis::PoolError::Timeout( - deadpool_redis::TimeoutType::Wait, - ))) - } - async fn begin_serving_write( &self, community_id: CommunityId, @@ -269,13 +248,13 @@ impl SessionDirectory { let serving_write = self.begin_serving_write(community_id).await?; let keys = SessionKeys::new(community_id, session_id); let ttl_ms = ttl_ms(self.lease_ttl)?; - let mut conn = self.connection().await?; + let mut conn = self.pool.clone(); let mutation = async { let reply = eval(ACQUIRE_SCRIPT, &keys) .arg(owner_runtime_id.to_hex()) .arg(profile.as_wire_str()) .arg(ttl_ms) - .query_async::<(String, String, String)>(&mut *conn) + .query_async::<(String, String, String)>(&mut conn) .await; #[cfg(test)] crate::nip_fi_test_hooks::after_directory_cas(community_id).await; @@ -327,14 +306,14 @@ impl SessionDirectory { /// the key expires with its TTL. async fn discard_unverified( &self, - conn: &mut deadpool_redis::Connection, + conn: &mut buzz_pubsub::RedisConnection, keys: &SessionKeys, lease: &SessionLease, ) { let discarded = eval(RELEASE_SCRIPT, keys) .arg(lease.owner_runtime_id.to_hex()) .arg(lease.generation) - .query_async::<(String, String, String)>(&mut **conn) + .query_async::<(String, String, String)>(&mut *conn) .await; if let Err(error) = discarded { tracing::warn!( @@ -369,13 +348,13 @@ impl SessionDirectory { let serving_write = self.begin_serving_write(lease.community_id).await?; let keys = SessionKeys::new(lease.community_id, lease.session_id); let ttl_ms = ttl_ms(self.lease_ttl)?; - let mut conn = self.connection().await?; + let mut conn = self.pool.clone(); let mutation = async { eval(RENEW_SCRIPT, &keys) .arg(lease.owner_runtime_id.to_hex()) .arg(lease.generation) .arg(ttl_ms) - .query_async(&mut *conn) + .query_async(&mut conn) .await }; let (status, value, known_generation): (String, String, String) = match &serving_write { @@ -413,12 +392,12 @@ impl SessionDirectory { pub async fn release(&self, lease: &SessionLease) -> Result { let serving_write = self.begin_serving_write(lease.community_id).await?; let keys = SessionKeys::new(lease.community_id, lease.session_id); - let mut conn = self.connection().await?; + let mut conn = self.pool.clone(); let mutation = async { eval(RELEASE_SCRIPT, &keys) .arg(lease.owner_runtime_id.to_hex()) .arg(lease.generation) - .query_async(&mut *conn) + .query_async(&mut conn) .await }; let (status, value, known_generation): (String, String, String) = match &serving_write { @@ -458,10 +437,10 @@ impl SessionDirectory { session_id: Uuid, ) -> Result, DirectoryError> { let keys = SessionKeys::new(community_id, session_id); - let mut conn = self.connection().await?; + let mut conn = self.pool.clone(); let value: Option = redis::cmd("GET") .arg(&keys.lease) - .query_async(&mut *conn) + .query_async(&mut conn) .await?; value .as_deref() @@ -476,10 +455,10 @@ impl SessionDirectory { session_id: Uuid, ) -> Result, DirectoryError> { let keys = SessionKeys::new(community_id, session_id); - let mut conn = self.connection().await?; + let mut conn = self.pool.clone(); let value: Option = redis::cmd("GET") .arg(&keys.generation) - .query_async(&mut *conn) + .query_async(&mut conn) .await?; match value.as_deref() { Some(value) => parse_optional_generation(community_id, session_id, value), @@ -500,12 +479,9 @@ impl SessionDirectory { fenced: &FencedHeader, ) -> Result<(), MeshError> { let keys = SessionKeys::new(community_id, fenced.session_id); - let mut conn = self - .connection() - .await - .map_err(|e| MeshError::Transport(format!("redis pool: {e}")))?; + let mut conn = self.pool.clone(); let (lease_value, known_generation): (String, String) = - eval(VALIDATE_SCRIPT, &keys).query_async(&mut *conn).await?; + eval(VALIDATE_SCRIPT, &keys).query_async(&mut conn).await?; let known_from_counter = parse_optional_generation(community_id, fenced.session_id, &known_generation) .map_err(|e| MeshError::Transport(e.to_string()))? @@ -735,18 +711,16 @@ mod tests { RuntimeId([byte; 32]) } - fn pool() -> deadpool_redis::Pool { + fn pool() -> buzz_pubsub::RedisConnection { let url = std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379".into()); - deadpool_redis::Config::from_url(url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("create redis pool") + buzz_pubsub::redis_connection(&url).expect("create redis pool") } async fn redis_directory_if_available() -> Option { let pool = pool(); - let mut conn = pool.get().await.ok()?; + let mut conn = pool.clone(); redis::cmd("PING") - .query_async::(&mut *conn) + .query_async::(&mut conn) .await .ok()?; Some(SessionDirectory::with_lease_ttl( @@ -757,11 +731,11 @@ mod tests { async fn clear_keys(directory: &SessionDirectory, community_id: CommunityId, session_id: Uuid) { let keys = SessionKeys::new(community_id, session_id); - let mut conn = directory.pool.get().await.expect("redis conn"); + let mut conn = directory.pool.clone(); let _: () = redis::cmd("DEL") .arg(keys.lease) .arg(keys.generation) - .query_async(&mut *conn) + .query_async(&mut conn) .await .expect("clear keys"); } diff --git a/crates/buzz-relay/src/tunnel/reliable.rs b/crates/buzz-relay/src/tunnel/reliable.rs index 8238d41083..9c465a2bd0 100644 --- a/crates/buzz-relay/src/tunnel/reliable.rs +++ b/crates/buzz-relay/src/tunnel/reliable.rs @@ -676,18 +676,16 @@ mod tests { RuntimeId([byte; 32]) } - fn pool() -> deadpool_redis::Pool { + fn pool() -> buzz_pubsub::RedisConnection { let url = std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379".into()); - deadpool_redis::Config::from_url(url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("create redis pool") + buzz_pubsub::redis_connection(&url).expect("create redis pool") } async fn redis_directory_if_available() -> Option { let pool = pool(); - let mut conn = pool.get().await.ok()?; + let mut conn = pool.clone(); redis::cmd("PING") - .query_async::(&mut *conn) + .query_async::(&mut conn) .await .ok()?; Some(SessionDirectory::with_lease_ttl( @@ -707,11 +705,11 @@ mod tests { profile: Profile::ReliableStream, }) .await; - let mut conn = pool().get().await.expect("redis conn"); + let mut conn = pool().clone(); let _: () = redis::cmd("DEL") .arg(format!("{base}:lease")) .arg(format!("{base}:generation")) - .query_async(&mut *conn) + .query_async(&mut conn) .await .expect("clear keys"); } diff --git a/crates/buzz-relay/src/workflow_sink.rs b/crates/buzz-relay/src/workflow_sink.rs index 9c80916cf6..4208b5b0e4 100644 --- a/crates/buzz-relay/src/workflow_sink.rs +++ b/crates/buzz-relay/src/workflow_sink.rs @@ -791,9 +791,7 @@ mod postgres_tests { config.redis_url = "redis://127.0.0.1:1".to_string(); let pool = sqlx::PgPool::connect_lazy(&config.database_url).expect("lazy pg pool"); let db = buzz_db::Db::from_pool(pool.clone()); - let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .expect("redis pool"); + let redis_pool = buzz_pubsub::redis_connection(&config.redis_url).expect("redis pool"); let pubsub = Arc::new( buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) .await diff --git a/crates/buzz-relay/tests/boot_lifecycle.rs b/crates/buzz-relay/tests/boot_lifecycle.rs index 481b3eaf5d..110542a044 100644 --- a/crates/buzz-relay/tests/boot_lifecycle.rs +++ b/crates/buzz-relay/tests/boot_lifecycle.rs @@ -716,7 +716,7 @@ mod postgres_tests { } /// Redis is required for pub/sub fan-out, presence, and typing, but nothing - /// in boot ever opened a command connection: `deadpool_redis` pools dial + /// in boot ever opened a command connection: the shared Redis connection dials /// lazily and `PubSubManager::new` only allocates channels, so "Redis /// pub/sub connected" was logged against a dead port. A relay could /// therefore boot with Redis unreachable, bind its health listener, and — @@ -774,7 +774,7 @@ mod postgres_tests { } /// A peer that accepts the socket but withholds its Redis response exercises - /// the outer timeout around both lazy pool checkout and PING. Removing or + /// the outer timeout around connect, handshake, and PING. Removing or /// narrowing that timeout makes this test kill a still-running relay at the /// deadline instead of observing a startup failure. #[test] diff --git a/scripts/run-tests.sh b/scripts/run-tests.sh index c822a33064..dc87f85af7 100755 --- a/scripts/run-tests.sh +++ b/scripts/run-tests.sh @@ -98,6 +98,11 @@ run_unit_tests() { run_test_step "buzz-pubsub conn_control tests" \ cargo test -p buzz-pubsub --lib conn_control::tests:: -- --nocapture + # Shared Redis connection timeout bounds (loopback fake peers; infra-free). + # Mirrors `just test-unit`. + run_test_step "buzz-pubsub connection tests" \ + cargo test -p buzz-pubsub --lib connection::tests:: -- --nocapture + run_test_step "buzz-voice tests" \ cargo test -p buzz-voice --lib -- --nocapture