diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index bc594e16ad..2c471d5a7a 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -353,6 +353,7 @@ jobs: cargo nextest archive \ --cargo-profile ci \ -p buzz-db \ + -p buzz-pubsub \ -p buzz-relay \ -p buzz-test-client \ --lib \ @@ -684,6 +685,22 @@ jobs: done cat /tmp/buzz-relay.log exit 1 + - name: Redis compatibility tests (standalone) + # buzz-pubsub's Redis-backed tests are #[ignore]d, so nothing ran them + # in CI before. They assert the cluster-safe command shapes required by + # ElastiCache Serverless (single-key GETs for bulk presence, exact + # SUBSCRIBE) *and* that those shapes still behave on standard Redis — + # which is the compatibility half of the guarantee, and the half a + # serverless-only rig cannot check. Points at the standalone Redis this + # job already runs; a cluster-mode service container is deliberately not + # added here. + run: | + cargo nextest run \ + --archive-file target/ci/backend-integration-tests.tar.zst \ + -E 'package(buzz-pubsub)' \ + --run-ignored all + env: + REDIS_URL: redis://localhost:6379 - name: Invite security tests run: | cargo nextest run \ diff --git a/crates/buzz-pubsub/src/error.rs b/crates/buzz-pubsub/src/error.rs index 96bc1016c9..6461f9b4b6 100644 --- a/crates/buzz-pubsub/src/error.rs +++ b/crates/buzz-pubsub/src/error.rs @@ -26,6 +26,17 @@ pub enum PubSubError { /// A Redis channel key could not be parsed as a valid channel ID. #[error("Invalid channel key: {0}")] InvalidChannelKey(String), + + /// A bulk presence read returned a different number of values than keys + /// requested. Reassembly is positional, so continuing would attribute one + /// pubkey's status to another. + #[error("Presence bulk read returned {got} values for {expected} keys")] + PresenceReplyMismatch { + /// Number of presence keys requested. + expected: usize, + /// Number of values Redis returned. + got: usize, + }, } impl From for PubSubError { diff --git a/crates/buzz-pubsub/src/lib.rs b/crates/buzz-pubsub/src/lib.rs index 4f1690beef..0cdd819456 100644 --- a/crates/buzz-pubsub/src/lib.rs +++ b/crates/buzz-pubsub/src/lib.rs @@ -368,8 +368,21 @@ impl PubSubManager { #[cfg(test)] pub(crate) mod test_util { + /// The Redis endpoint under test. Honours `REDIS_URL` so the same + /// `--ignored` suite can be pointed at standalone Redis (the CI default) or + /// at a single-shard cluster-mode server, which is what enforces + /// `CROSSSLOT` locally. + /// + /// Every test that opens a connection must route through this — a + /// hardcoded URL alongside an env-var-driven one makes the pool and the + /// pub/sub client talk to *different servers*, which fails as a timeout and + /// reads like a cluster incompatibility. + pub fn test_redis_url() -> String { + std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379".to_string()) + } + pub fn make_test_pool() -> deadpool_redis::Pool { - let cfg = deadpool_redis::Config::from_url("redis://127.0.0.1:6379"); + let cfg = deadpool_redis::Config::from_url(test_redis_url()); cfg.create_pool(Some(deadpool_redis::Runtime::Tokio1)) .expect("Failed to create Redis pool") } @@ -378,7 +391,7 @@ pub(crate) mod test_util { #[cfg(test)] mod tests { use super::*; - use crate::test_util::make_test_pool; + use crate::test_util::{make_test_pool, test_redis_url}; use buzz_core::{CommunityId, TenantContext}; use nostr::{EventBuilder, Keys, Kind}; use uuid::Uuid; @@ -386,7 +399,7 @@ mod tests { async fn make_manager() -> Arc { let pool = make_test_pool(); Arc::new( - PubSubManager::new("redis://127.0.0.1:6379", pool) + PubSubManager::new(&test_redis_url(), pool) .await .expect("Failed to create PubSubManager"), ) @@ -513,7 +526,7 @@ mod tests { let pool = make_test_pool(); let manager = Arc::new( PubSubManager::with_config( - PubSubConfig::new("redis://127.0.0.1:6379") + PubSubConfig::new(test_redis_url()) .with_unsubscribe_debounce(Duration::from_millis(25)), pool, ) @@ -593,8 +606,7 @@ mod tests { async fn retain_release_refcounts_and_debounces_last_release() { let pool = make_test_pool(); let manager = PubSubManager::with_config( - PubSubConfig::new("redis://127.0.0.1:6379") - .with_unsubscribe_debounce(Duration::from_millis(1)), + PubSubConfig::new(test_redis_url()).with_unsubscribe_debounce(Duration::from_millis(1)), pool, ) .await diff --git a/crates/buzz-pubsub/src/presence.rs b/crates/buzz-pubsub/src/presence.rs index e0c9dfd6c9..a5a357d488 100644 --- a/crates/buzz-pubsub/src/presence.rs +++ b/crates/buzz-pubsub/src/presence.rs @@ -71,6 +71,18 @@ pub async fn get_presence( } /// Returns `pubkey_hex → status` for all currently-set keys. +/// +/// Issues one single-key `GET` per pubkey in a **non-atomic** pipeline: presence +/// keys are per-pubkey, so on a cluster (ElastiCache Serverless included) they +/// span slots and any multi-key command over them fails `CROSSSLOT`. +/// +/// **Never call `.atomic()` on this pipeline.** `Pipeline::new` leaves +/// `transaction_mode: false`; `.atomic()` sets it, which makes `query_async` +/// wrap the batch in `MULTI`/`EXEC` and dispatch it as one unit — restoring +/// exactly the cross-slot failure this replaces. A non-atomic pipeline is +/// independent commands sharing one round trip, which is all this read needs: +/// each key is fetched on its own and the entries are TTL'd anyway, so there is +/// nothing for atomicity to protect. pub async fn get_presence_bulk( pool: &Pool, ctx: &TenantContext, @@ -80,11 +92,22 @@ pub async fn get_presence_bulk( return Ok(HashMap::new()); } let mut conn = pool.get().await?; - let keys: Vec = pubkeys - .iter() - .map(|pubkey| presence_key(ctx, pubkey)) - .collect(); - let values: Vec> = redis::cmd("MGET").arg(&keys).query_async(&mut conn).await?; + let mut pipe = redis::pipe(); + for pubkey in pubkeys { + pipe.cmd("GET").arg(presence_key(ctx, pubkey)); + } + let values: Vec> = pipe.query_async(&mut conn).await?; + + // Replies are positional — Redis returns values, not key names — so a reply + // of the wrong length would silently misattribute one pubkey's status to + // another under `zip`. Refuse instead. + if values.len() != pubkeys.len() { + return Err(PubSubError::PresenceReplyMismatch { + expected: pubkeys.len(), + got: values.len(), + }); + } + let result = pubkeys .iter() .zip(values.iter()) @@ -139,6 +162,49 @@ mod tests { ); } + /// CRC16-XMODEM over the cluster hash-tag-free key, mod 16384 — the Redis + /// Cluster slot function. Test-local on purpose: production never computes + /// slots (we use a plain, non-cluster client), this exists only to assert a + /// key-design property. + fn key_slot(key: &str) -> u16 { + let mut crc: u16 = 0; + for byte in key.as_bytes() { + crc ^= (*byte as u16) << 8; + for _ in 0..8 { + crc = if crc & 0x8000 != 0 { + (crc << 1) ^ 0x1021 + } else { + crc << 1 + }; + } + } + crc % 16384 + } + + #[test] + fn key_slot_matches_redis_cluster_reference_vector() { + // Redis Cluster spec's CRC16 check value: CRC16-XMODEM("123456789") = + // 0x31C3, which is below 16384 so the slot equals it outright. + assert_eq!(key_slot("123456789"), 0x31C3); + } + + /// Presence keys carry no hash tag, so pubkeys in one community spread across + /// slots. That is deliberate: tagging them by community would pin an entire + /// community's presence to a single slot, against AWS's uniform-distribution + /// guidance and into the per-slot ECPU ceiling. It is also why the bulk read + /// must be single-key `GET`s rather than any multi-key command. + #[test] + fn presence_keys_in_one_community_span_multiple_slots() { + let ctx = ctx(0xaaaa, "a.example"); + let slots: std::collections::HashSet = (0..16) + .map(|_| key_slot(&presence_key(&ctx, &make_pubkey()))) + .collect(); + assert!( + slots.len() > 1, + "presence keys must not co-slot; got {slots:?}" + ); + } + #[tokio::test] #[ignore = "requires Redis"] async fn test_presence_set_and_get() { @@ -189,6 +255,91 @@ mod tests { clear_presence(&pool, &ctx, &pk2).await.unwrap(); } + /// Asserts the **reassembled map**, not the reply vector or its length. + /// + /// The bulk read zips request positions against reply positions, so a + /// misalignment is count-correct and attributes one member's status to + /// another — silent server-side, and user-visible as other people's online + /// status. Every assertion here is `pubkey → status` for a *distinctly + /// valued* status, so any permutation of the reply fails. + #[tokio::test] + #[ignore = "requires Redis"] + async fn presence_bulk_reassembles_by_pubkey_across_slots() { + let pool = make_test_pool(); + let ctx = ctx(0xb301, "b3.example"); + + // Distinct statuses so a shifted reply can't coincidentally match, and + // enough keys that they span multiple slots (see the slot unit test). + let present: Vec<(PublicKey, String)> = (0..12) + .map(|i| (make_pubkey(), format!("status-{i}"))) + .collect(); + let missing: Vec = (0..4).map(|_| make_pubkey()).collect(); + + for (pk, status) in &present { + set_presence(&pool, &ctx, pk, status).await.unwrap(); + } + + let expected: HashMap = present + .iter() + .map(|(pk, status)| (pk.to_hex(), status.clone())) + .collect(); + + // Interleave set and unset keys, so a `None` in the middle of the reply + // must not shift the keys after it. + let mut requested: Vec = Vec::new(); + for (i, (pk, _)) in present.iter().enumerate() { + requested.push(*pk); + if let Some(absent) = missing.get(i / 3) { + requested.push(*absent); + } + } + + let got = get_presence_bulk(&pool, &ctx, &requested).await.unwrap(); + assert_eq!(got, expected, "reassembled map must be keyed by pubkey"); + + // Same set, reversed order: the map is order-independent by definition, + // so this catches positional misalignment without needing to know which + // order is "right". + let mut reversed = requested.clone(); + reversed.reverse(); + let got_reversed = get_presence_bulk(&pool, &ctx, &reversed).await.unwrap(); + assert_eq!( + got_reversed, expected, + "map must not depend on request order" + ); + + // Duplicate keys: the reply is one value per request position, and the + // map must collapse them rather than drift. + let duplicated: Vec = + requested.iter().chain(requested.iter()).copied().collect(); + let got_duplicated = get_presence_bulk(&pool, &ctx, &duplicated).await.unwrap(); + assert_eq!( + got_duplicated, expected, + "duplicate requests must collapse to the same map" + ); + + // All-missing: no entries, no error. + assert!(get_presence_bulk(&pool, &ctx, &missing) + .await + .unwrap() + .is_empty()); + + for (pk, _) in &present { + clear_presence(&pool, &ctx, pk).await.unwrap(); + } + } + + #[tokio::test] + #[ignore = "requires Redis"] + async fn presence_bulk_empty_request_returns_empty_map() { + let pool = make_test_pool(); + let ctx = ctx(0xb302, "b3.example"); + assert!(get_presence_bulk(&pool, &ctx, &[]) + .await + .unwrap() + .is_empty()); + } + #[tokio::test] #[ignore = "requires Redis"] async fn test_presence_ttl() {