fix(pubsub): replace cross-slot presence MGET with non-atomic pipeline

Bulk presence read issued a single `MGET` over one key per pubkey. Presence
keys are `buzz:{community}:presence:{pubkey_hex}` with no hash tag, so they
hash to different cluster slots and the command fails `CROSSSLOT` on any
cluster-mode server, including ElastiCache Serverless.

Replace it with one single-key `GET` per pubkey in a non-atomic pipeline:
same one round trip, same key format, works identically on standalone Redis.
The key format is deliberately left untagged — tagging presence by community
would co-slot an entire community's keys and concentrate load on one slot.

Reply reassembly is positional, so a short reply would silently attribute one
member's status to another (user-visible as other people's online status).
Length is now checked and mismatches error instead of truncating under `zip`.

`.atomic()` must never be added to this pipeline: it wraps the batch in
MULTI/EXEC, which is dispatched as a unit and restores the cross-slot failure.
Documented at the call site, since single-node tests cannot catch it.

Also route the test Redis endpoint through one `test_redis_url()` helper.
`make_test_pool()` hardcoded `127.0.0.1:6379` while pub/sub tests passed their
own literal URL, so pointing the suite at another server split the pool and the
pub/sub client across two hosts — which surfaced as three unrelated timeouts
that looked like cluster incompatibility.

Tests: bulk read asserts the reassembled `pubkey -> status` map (not vector
length) across distinct slots, shuffled order, interleaved missing keys, and
duplicates, plus permutation-equality so misalignment fails without knowing
the expected order. A test-local CRC16-XMODEM (checked against the Redis
Cluster reference vector) asserts presence keys span multiple slots, which
also fails if someone later hash-tags them.

CI now runs the buzz-pubsub suite including `#[ignore]`d Redis tests against
the standalone Redis the backend-integration job already starts; previously
nothing ran them. This keeps standard Redis in the matrix permanently.

Verified at this tree: 39/39 with `--include-ignored` against standalone
Redis 7 and against a single-shard cluster-mode server owning all 16384
slots. Negative control: restoring the `MGET` on the same rig fails with
`CrossSlot: Keys in request don't hash to the same slot`, confirming the rig
detects the bug the fix removes.

Co-authored-by: Dawn (sprout agent) <c6237ef84fa537c78dcee78efd2d4e59f728859c7f194da42ac51ededfa0be05@sprout-oss.stage.blox.sqprod.co>
Signed-off-by: tlongwell-block <109685178+tlongwell-block@users.noreply.github.com>

.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 \

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<tokio::sync::broadcast::error::RecvError> for PubSubError {

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<PubSubManager> {
         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

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<String> = pubkeys
-        .iter()
-        .map(|pubkey| presence_key(ctx, pubkey))
-        .collect();
-    let values: Vec<Option<String>> = 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<Option<String>> = 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<u16> = (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<PublicKey> = (0..4).map(|_| make_pubkey()).collect();
+
+        for (pk, status) in &present {
+            set_presence(&pool, &ctx, pk, status).await.unwrap();
+        }
+
+        let expected: HashMap<String, String> = 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<PublicKey> = 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<PublicKey> =
+            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() {