Traffic Skew with Centrifugo's Sharded Pub/Sub in Redis Cluster

[allOwO]

I'd like to share an observation regarding Redis Cluster traffic skew when using Centrifugo's Sharded Pub/Sub, and ask for some advice.

Context & Observation

When load testing Sharded Pub/Sub with 32 partitions, traffic distributes well on a 4-shard Redis Cluster. However, scaling up reveals severe traffic skew due to how Redis CRC16 hashes short sequential partition indices:

Shards	Configured Partitions	Observation
4	32	Balanced
8	32	Highly skewed (4 out of 8 shards idle)
16	32	Highly skewed (8 out of 16 shards idle)

The Root Cause

In broker_redis.go, the Sharded Pub/Sub partition index is cast to a string and used directly as the Redis hash tag:

// builder appends: "prefix{" + idxStr + "}." + ch
builder.WriteString(idxStr)
// ...

Since Redis slot calculation depends entirely on the string inside {} (i.e. CRC16("0") to CRC16("31")), it leads to massive collisions on larger clusters.

A quick Python simulation (code at the bottom) confirms that half the shards receive 0 keys when scaling up.

16-shard distribution result snippet:

Shard 0 : 3 keys    Shard 8 : 2 keys
Shard 1 : 5 keys    Shard 9 : 6 keys
Shard 4 : 3 keys    Shard 12: 2 keys
Shard 5 : 5 keys    Shard 13: 6 keys
Shard 2,3,6,7,10,11,14,15: 0 keys (idle)

Looking for Advice

To workaround this, we are considering:

Approach	Example	Rationale
Explicit Prefix	{pt0}ch	Using custom hash tags avoids partition index skew.

Would introducing configurable explicit prefixes (acting as custom hash tags) be a good practice or an acceptable upstream feature? Thanks in advance for your insights!

---

Full Python Code & Results

crc16tab = [
    0x0000,0x1021,0x2042,0x3063,0x4084,0x50a5,0x60c6,0x70e7,
    0x8108,0x9129,0xa14a,0xb16b,0xc18c,0xd1ad,0xe1ce,0xf1ef,
    0x1231,0x0210,0x3273,0x2252,0x52b5,0x4294,0x72f7,0x62d6,
    0x9339,0x8318,0xb37b,0xa35a,0xd3bd,0xc39c,0xf3ff,0xe3de,
    0x2462,0x3443,0x0420,0x1401,0x64e6,0x74c7,0x44a4,0x5485,
    0xa56a,0xb54b,0x8528,0x9509,0xe5ee,0xf5cf,0xc5ac,0xd58d,
    0x3653,0x2672,0x1611,0x0630,0x76d7,0x66f6,0x5695,0x46b4,
    0xb75b,0xa77a,0x9719,0x8738,0xf7df,0xe7fe,0xd79d,0xc7bc,
    0x48c4,0x58e5,0x6886,0x78a7,0x0840,0x1861,0x2802,0x3823,
    0xc9cc,0xd9ed,0xe98e,0xf9af,0x8948,0x9969,0xa90a,0xb92b,
    0x5af5,0x4ad4,0x7ab7,0x6a96,0x1a71,0x0a50,0x3a33,0x2a12,
    0xdbfd,0xcbdc,0xfbbf,0xeb9e,0x9b79,0x8b58,0xbb3b,0xab1a,
    0x6ca6,0x7c87,0x4ce4,0x5cc5,0x2c22,0x3c03,0x0c60,0x1c41,
    0xedae,0xfd8f,0xcdec,0xddcd,0xad2a,0xbd0b,0x8d68,0x9d49,
    0x7e97,0x6eb6,0x5ed5,0x4ef4,0x3e13,0x2e32,0x1e51,0x0e70,
    0xff9f,0xefbe,0xdfdd,0xcffc,0xbf1b,0xaf3a,0x9f59,0x8f78,
    0x9188,0x81a9,0xb1ca,0xa1eb,0xd10c,0xc12d,0xf14e,0xe16f,
    0x1080,0x00a1,0x30c2,0x20e3,0x5004,0x4025,0x7046,0x6067,
    0x83b9,0x9398,0xa3fb,0xb3da,0xc33d,0xd31c,0xe37f,0xf35e,
    0x02b1,0x1290,0x22f3,0x32d2,0x4235,0x5214,0x6277,0x7256,
    0xb5ea,0xa5cb,0x95a8,0x8589,0xf56e,0xe54f,0xd52c,0xc50d,
    0x34e2,0x24c3,0x14a0,0x0481,0x7466,0x6447,0x5424,0x4405,
    0xa7db,0xb7fa,0x8799,0x97b8,0xe75f,0xf77e,0xc71d,0xd73c,
    0x26d3,0x36f2,0x0691,0x16b0,0x6657,0x7676,0x4615,0x5634,
    0xd94c,0xc96d,0xf90e,0xe92f,0x99c8,0x89e9,0xb98a,0xa9ab,
    0x5844,0x4865,0x7806,0x6827,0x18c0,0x08e1,0x3882,0x28a3,
    0xcb7d,0xdb5c,0xeb3f,0xfb1e,0x8bf9,0x9bd8,0xabbb,0xbb9a,
    0x4a75,0x5a54,0x6a37,0x7a16,0x0af1,0x1ad0,0x2ab3,0x3a92,
    0xfd2e,0xed0f,0xdd6c,0xcd4d,0xbdaa,0xad8b,0x9de8,0x8dc9,
    0x7c26,0x6c07,0x5c64,0x4c45,0x3ca2,0x2c83,0x1ce0,0x0cc1,
    0xef1f,0xff3e,0xcf5d,0xdf7c,0xaf9b,0xbfba,0x8fd9,0x9ff8,
    0x6e17,0x7e36,0x4e55,0x5e74,0x2e93,0x3eb2,0x0ed1,0x1ef0
]

def redis_crc16(s):
    crc = 0
    for b in s.encode('utf-8'):
        crc = ((crc << 8) ^ crc16tab[((crc >> 8) ^ b) & 0x00FF]) & 0xFFFF
    return crc

shard_counts_4 = {i: 0 for i in range(4)}
shard_counts_8 = {i: 0 for i in range(8)}
shard_counts_16 = {i: 0 for i in range(16)}

print(f"{'Key':<8} | {'Node 4-Shard':<14} | {'Node 8-Cluster':<14} | {'Node 16-Shard':<14}")
print('-' * 60)
for i in range(32):
    slot = redis_crc16(str(i)) % 16384
    
    # 4 Shard: each chunk is 4096 slots
    shard_4 = slot // 4096
    shard_counts_4[shard_4] += 1
    
    # 8 Shard: each chunk is 2048 slots
    shard_8 = slot // 2048
    shard_counts_8[shard_8] += 1
    
    # 16 Shard: each chunk is 1024 slots
    shard_16 = slot // 1024
    shard_counts_16[shard_16] += 1
    
    print(f"{i:<8} | {shard_4:<14} | {shard_8:<14} | {shard_16:<14}")

print('\nDistribution across 16 shards:')
for shard, count in shard_counts_16.items():
    print(f'Shard {shard:<2}: {count} keys')

Output:

Key      | Node 4-Shard   | Node 8-Shard   | Node 16-Shard 
------------------------------------------------------------
0        | 3              | 6              | 13            
1        | 2              | 4              | 9             
2        | 1              | 2              | 5             
3        | 0              | 0              | 1             
4        | 3              | 6              | 13            
5        | 2              | 4              | 9             
6        | 1              | 2              | 5             
7        | 0              | 0              | 1             
8        | 3              | 6              | 13            
9        | 2              | 4              | 9             
10       | 0              | 0              | 0             
11       | 1              | 2              | 4             
12       | 2              | 4              | 8             
13       | 3              | 6              | 12            
14       | 0              | 0              | 0             
15       | 1              | 2              | 4             
16       | 2              | 4              | 8             
17       | 3              | 6              | 12            
18       | 0              | 0              | 0             
19       | 1              | 2              | 4             
20       | 1              | 2              | 5             
21       | 0              | 0              | 1             
22       | 3              | 6              | 13            
23       | 2              | 4              | 9             
24       | 1              | 2              | 5             
25       | 0              | 0              | 1             
26       | 3              | 6              | 13            
27       | 2              | 4              | 9             
28       | 1              | 2              | 5             
29       | 0              | 0              | 1             
30       | 2              | 4              | 9             
31       | 3              | 6              | 13            

Distribution across 16 shards:
Shard 0 : 3 keys
Shard 1 : 5 keys
Shard 2 : 0 keys
Shard 3 : 0 keys
Shard 4 : 3 keys
Shard 5 : 5 keys
Shard 6 : 0 keys
Shard 7 : 0 keys
Shard 8 : 2 keys
Shard 9 : 6 keys
Shard 10: 0 keys
Shard 11: 0 keys
Shard 12: 2 keys
Shard 13: 6 keys
Shard 14: 0 keys
Shard 15: 0 keys

[FZambia]

Hello @allOwO! Thanks for the detailed report.

May I ask why do you experiment with only 32 channels? It's seems a very small number to expect an equal distribution. If you try to test this out on 32k channels instead of 32 – then you will get more fair distribution.

[allOwO]

Hi @FZambia, thanks for the quick response!

I think there might be a small misunderstanding — the skew is not about channel count, but about the partition hash tags themselves.

In messageChannelID() (broker_redis.go):

// Sharded pub/sub path: uses partition-based hash tags
idx := consistentIndex(ch, b.config.NumShardedPubSubPartitions)
idxStr := strconv.Itoa(idx)

// Pre-calculate capacity: messagePrefix + "{" + idxStr + "}." + ch
capacity := len(b.messagePrefix) + 1 + len(idxStr) + 2 + len(ch)

var builder strings.Builder
builder.Grow(capacity)
builder.WriteString(b.messagePrefix)
builder.WriteByte('{')
builder.WriteString(idxStr)
builder.WriteString("}.")
builder.WriteString(ch)

Redis determines the slot solely from the content inside {}, so regardless of how many channels exist, all channels within partition N share hash tag {N} and land on the same slot. The documentation recommends 64 to 128 partitions — 128 does produce a better spread, but CRC16 over short sequential digit strings still doesn't distribute uniformly across the 16384-slot space, causing noticeable imbalance on larger clusters.

One simple fix that works well in practice: add a short prefix to the hash tag (e.g. {pt0} instead of {0}). In my testing, a "pt" prefix produces the most even distribution — for example, 128 partitions on a 16-shard cluster goes from 5–10 spread to 7–10, and no idle shards even at 32 shards. The change is minimal and backward-compatible if introduced as a new option.

Would you be open to this direction? Happy to contribute a PR.

[FZambia]

Ah, my bad, I see now, I was thinking about channels to partition number distribution.

A side question to understand the use case - in your case, why you decided to go with 32 partitions, instead of 128 for example? To reduce the number of connections? Or some other reason?

I suppose the prefix idea will improve situation, but not fully. It's still hard to find single proper prefix.

What do you think about the following approach: we can generate stable partition identifiers based on partition count and optimize it for max cluster size. Please take a look at the attached Python program.

In that case I suppose it would be better to provide an option to set ShardedPubSubPartitionTags []string, instead of prefix. So that users could provide any precomputed distribution. In that case NumShardedPubSubPartitions is not needed at all since the number is len(ShardedPubSubPartitionTags) (and Centrifuge should return an error if both provided in the configuration).

"""
Find NUM_PARTITIONS partition tags with even distribution across cluster sizes 1-MAX_CLUSTER,
with graceful degradation for larger clusters up to any size.
"""
import binascii, itertools, random, math

TOTAL_SLOTS = 16384
NUM_PARTITIONS = 64
MAX_CLUSTER = 64

def redis_crc16(data: bytes) -> int:
    return binascii.crc_hqx(data, 0) & 0xFFFF

def slot_to_node(slot, num_nodes):
    """O(1) node lookup, works for ANY cluster size."""
    sn = TOTAL_SLOTS // num_nodes
    r = TOTAL_SLOTS % num_nodes
    b = r * (sn + 1)
    return slot // (sn + 1) if slot < b else r + (slot - b) // sn

# ─── Precompute for annealing (sizes 1..MAX_CLUSTER only) ──────────
print("Precomputing node mappings...")
NODE_OF = [None]
for k in range(1, MAX_CLUSTER + 1):
    sn = TOTAL_SLOTS // k
    r = TOTAL_SLOTS % k
    b = r * (sn + 1)
    m = [0] * TOTAL_SLOTS
    for s in range(TOTAL_SLOTS):
        m[s] = s // (sn + 1) if s < b else r + (s - b) // sn
    NODE_OF.append(m)

TARGETS = [None]
for k in range(1, MAX_CLUSTER + 1):
    lo = NUM_PARTITIONS // k
    TARGETS.append((lo, lo + (1 if NUM_PARTITIONS % k else 0)))
print("Done.\n")

def viol(c, lo, hi):
    if c > hi: return c - hi
    if c < lo: return lo - c
    return 0

def full_evaluate(counts_per_k):
    v = 0
    for k in range(1, MAX_CLUSTER + 1):
        lo, hi = TARGETS[k]
        for n in range(k):
            v += viol(counts_per_k[k][n], lo, hi)
    return v

def build_counts(slots):
    counts = [None]
    for k in range(1, MAX_CLUSTER + 1):
        arr = [0] * k
        for s in slots:
            arr[NODE_OF[k][s]] += 1
        counts.append(arr)
    return counts

def compute_delta(counts, old_slot, new_slot):
    d = 0
    for k in range(1, MAX_CLUSTER + 1):
        on = NODE_OF[k][old_slot]
        nn = NODE_OF[k][new_slot]
        if on == nn:
            continue
        lo, hi = TARGETS[k]
        co, cn = counts[k][on], counts[k][nn]
        d += (viol(co - 1, lo, hi) + viol(cn + 1, lo, hi)
            - viol(co, lo, hi) - viol(cn, lo, hi))
    return d

def apply(counts, old_slot, new_slot):
    for k in range(1, MAX_CLUSTER + 1):
        on = NODE_OF[k][old_slot]
        nn = NODE_OF[k][new_slot]
        if on != nn:
            counts[k][on] -= 1
            counts[k][nn] += 1

# ─── Phase 1: Simulated Annealing ──────────────────────────────────

def find_balanced_slots(seed=42, iters=5_000_000):
    random.seed(seed)
    slots = random.sample(range(TOTAL_SLOTS), NUM_PARTITIONS)
    slot_set = set(slots)
    counts = build_counts(slots)
    score = full_evaluate(counts)
    best, best_slots = score, slots[:]
    print(f"Initial violation: {score}")

    temp, cool = 60.0, 0.999975
    for it in range(iters):
        idx = random.randint(0, NUM_PARTITIONS - 1)
        old = slots[idx]
        new = random.randint(0, TOTAL_SLOTS - 1)
        if new in slot_set:
            continue

        d = compute_delta(counts, old, new)

        if d <= 0 or random.random() < math.exp(-d / max(temp, .001)):
            apply(counts, old, new)
            slot_set.discard(old); slot_set.add(new)
            slots[idx] = new
            score += d
            if score < best:
                best, best_slots = score, slots[:]
                if best == 0:
                    print(f"  Perfect at iteration {it:,}")
                    return best_slots
        temp *= cool
        if it % 500_000 == 0:
            print(f"  {it:>9,}: temp={temp:.3f} score={score} best={best}")

    print(f"  Best: {best}")
    return best_slots

# ─── Phase 2: Find String Tags ─────────────────────────────────────

def find_tags(target_slots):
    m, needed = {}, set(target_slots)
    chars = "abcdefghijklmnopqrstuvwxyz0123456789"
    for ln in range(2, 8):
        if not needed: break
        print(f"  len={ln}: {len(needed)} left")
        for c in itertools.product(chars, repeat=ln):
            t = "".join(c)
            s = redis_crc16(t.encode()) % TOTAL_SLOTS
            if s in needed and s not in m:
                m[s] = t; needed.discard(s)
                if not needed: break
    return m

# ─── Phase 3: Verification & Distribution ──────────────────────────

def show_distribution(tags, tag_slots, cluster_size):
    print(f"\n{'=' * 60}")
    print(f"Cluster size: {cluster_size} nodes")
    print(f"{'=' * 60}")

    lo = NUM_PARTITIONS // cluster_size
    hi = lo + (1 if NUM_PARTITIONS % cluster_size else 0)
    print(f"Expected partitions per node: {lo}" +
          (f" or {hi}" if hi != lo else ""))

    # print(f"\n{'Idx':>3}  {'Tag':>8}  {'Slot':>6}  {'Node':>5}")
    # print("-" * 30)

    node_partitions = [[] for _ in range(cluster_size)]
    for i, tag in enumerate(tags):
        s = tag_slots[tag]
        n = slot_to_node(s, cluster_size)
        node_partitions[n].append(tag)
        # print(f"{i:3}  {tag:>8}  {s:6}  {n:5}")

    counts = [len(p) for p in node_partitions]
    mn, mx = min(counts), max(counts)
    active = sum(1 for c in counts if c > 0)
    empty = cluster_size - active

    print(f"\nPartitions per node:")
    print("-" * 40)
    for n in range(cluster_size):
        c = counts[n]
        if c > 0 or cluster_size <= 40:
            bar = "█" * c
            tags_str = ", ".join(node_partitions[n]) if c > 0 else ""
            print(f"  Node {n:2}: {c:2} {bar}  {tags_str}")

    if empty > 0 and cluster_size > 40:
        print(f"  ... {empty} nodes with 0 partitions (expected)")

    print(f"\n  Active nodes: {active}/{cluster_size}")
    print(f"  Range: [{mn},{mx}]")
    balanced = mx - mn <= 1
    degraded = cluster_size > NUM_PARTITIONS
    print(f"  Balanced: {'✓' if balanced else '✗'}" +
          (" (degraded — more nodes than partitions)" if degraded else ""))


# ─── Main ───────────────────────────────────────────────────────────

if __name__ == "__main__":
    print("=" * 60)
    print("Phase 1: Simulated annealing for balanced slots")
    print("=" * 60)
    slots = sorted(find_balanced_slots())
    print(f"\nSlots: {slots}")

    print("\n" + "=" * 60)
    print("Phase 2: Finding short string tags")
    print("=" * 60)
    tags_map = find_tags(slots)

    # Verify balance for optimized range
    print("\n" + "=" * 60)
    print("Balance verification (optimized range 1-32)")
    print("=" * 60)
    all_ok = True
    for k in range(1, MAX_CLUSTER + 1):
        lo, hi = TARGETS[k]
        cnts = [0] * k
        for s in slots:
            cnts[NODE_OF[k][s]] += 1
        mn, mx = min(cnts), max(cnts)
        ok = mx - mn <= 1
        if not ok: all_ok = False
        print(f"  {'OK' if ok else '!!'} k={k:2}: [{mn},{mx}] target [{lo},{hi}]")
    print(f"\n  All balanced (1-{MAX_CLUSTER}): {all_ok}")

    # Extended check beyond MAX_CLUSTER
    print("\n" + "=" * 60)
    print("Balance check (extended range 33-64)")
    print("=" * 60)
    for k in [33, 40, 48, 50, 64]:
        lo = NUM_PARTITIONS // k
        hi = lo + (1 if NUM_PARTITIONS % k else 0)
        cnts = [0] * k
        for s in slots:
            cnts[slot_to_node(s, k)] += 1
        mn, mx = min(cnts), max(cnts)
        ok = mx - mn <= 1
        active = sum(1 for c in cnts if c > 0)
        print(f"  {'OK' if ok else '!!'} k={k:2}: [{mn},{mx}] target [{lo},{hi}] "
              f"active={active}/{k}")

    # Output constant
    ordered_tags = [tags_map[s] for s in slots]
    tag_to_slot = {tags_map[s]: s for s in slots}

    print("\n" + "=" * 60)
    print("PARTITION_TAGS constant")
    print("=" * 60)
    print("PARTITION_TAGS = [")
    for s in slots:
        print(f'    "{tags_map[s]}",  # slot {s}')
    print("]")

    # Show full distribution for selected cluster sizes
    print("\n" + "=" * 60)
    print("Phase 3: Distribution examples")
    print("=" * 60)
    for cs in [2, 3, 4, 5, 6, 7, 8, 10, 16, 20, 25, 32, 40, 50, 64]:
        show_distribution(ordered_tags, tag_to_slot, cs)

[allOwO]

Thanks for the proposal — I think we can keep this simple and still flexible.

My preference is:

• provide built-in deterministic partition-tag mappings for 64 and 128,

• use 128 as the default,

• and do not expose ShardedPubSubPartitionTags []string as a user-facing config for now.

Why:

1. 128 is a good default for distribution quality.

2. 64 is still useful for smaller setups / lower connection overhead.

3. Built-in deterministic mappings keep behavior stable across nodes and restarts.

4. Avoiding []string config keeps UX simple and reduces misconfiguration risk.

Also, to clarify: the “32 partitions” in my earlier comment was only for fast test reproduction, not a production recommendation.

[FZambia]

Thx! Let's improve a bit further.

I just made a local Go package which pre-computes partitions (well, not me, but Opus 4.6). I suppose we can have it as internal package here. It will expose pre-computed partition tags for 16, 32, 64 ... 4096 partitions. We should then have a separate boolean option to use precomputed partitions to avoid breaking changes. And the option will require using the number of partitions which were pre-computed, error otherwise.

Does this work @allOwO ?

[allOwO]

@FZambia Thanks, this works for me.

I agree this is a better direction than exposing ShardedPubSubPartitionTags []string as a user-facing option. An opt-in boolean for precomputed partition tags keeps backward compatibility, keeps the configuration simpler, and still addresses the skew problem.
Requiring one of the supported precomputed partition counts also sounds reasonable to me. Happy to help test it or contribute a PR once the package/API shape is settled.

[FZambia]

Thanks @allOwO , I will push changes soon as I have the prebuilt partitions locally already and will ask you to take a look.

[FZambia]

Remember about this, just a huge scope of https://x.com/centrifugalabs/status/2041850361271451975 prevents me from doing this, but will do soon as this improvement is super useful.