Redis Internals: Pub/Sub Architecture and Data Structures

Redis Pub/Sub is an in-process fan-out over named channels and glob patterns. It is not the keyspace: messages are not stored as keys, have no TTL or RDB/AOF payload of their own, and are delivered only to clients that are subscribed at publish time. Delivery is best-effort (no server retry across disconnect). This post maps the structures, command paths, and delivery/security properties in git/redis (pubsub.c, fields on redisServer / client).

Related: Network / command path, Data types and encodings, Cluster bus gossip, Build from source.

1. Overview

A client in Pub/Sub mode (RESP2) may only run subscription control commands plus PING / QUIT / RESET until it unsubscribes from everything. In a cluster, a subscriber is visible solely on the node that accepted its connection; other nodes learn nothing about that client until a published payload arrives and they deliver to whoever is subscribed locally.

flowchart LR
  Pub["PUBLISH / SPUBLISH"] --> Core["pubsubPublishMessageInternal"]
  Core --> Ch["pubsub_channels / pubsubshard_channels"]
  Core --> Pat["pubsub_patterns"]
  Ch --> SA["addReplyPubsubMessage"]
  Pat --> SB["addReplyPubsubPatMessage"]
  SA --> Out["subscriber reply buffers"]
  SB --> Out

2. Server implementation

Pub/Sub does not store messages. The server keeps who is listening in forward indexes on redisServer (channel or pattern → set of client*), installs those edges on subscribe, and on PUBLISH / SPUBLISH walks the matching sets to addReply* into each subscriber’s output buffer. Global and shard paths share pubsubtype; after local fan-out, cluster or replication may propagate the payload so peer processes run the same local publish against their own indexes.

/* server.h — redisServer */
kvstore *pubsub_channels;       /* channel → dict of client* */
dict *pubsub_patterns;          /* pattern → dict of client* */
kvstore *pubsubshard_channels;  /* per-slot channel → dict of client* */
unsigned int pubsub_clients;    /* clients with CLIENT_PUBSUB */

@startuml
!option handwritten true
skinparam class {
    BackgroundColor White
    BorderColor Black
    ArrowColor Black
}

struct redisServer {
  +pubsub_channels : kvstore*
  +pubsub_patterns : dict*
  +pubsubshard_channels : kvstore*
  +pubsub_clients : uint
}

struct client {
  +pubsub_channels : dict*
  +pubsub_patterns : dict*
  +pubsubshard_channels : dict*
  +flags : uint64_t
}

struct "dict (subscribers)" as SubDict {
  key : client*
  val : NULL
}

struct "dict (membership)" as MembDict {
  key : robj* channel_or_pattern
  val : NULL
}

redisServer --> SubDict : per channel / pattern
client --> MembDict : channels / patterns this client joined
SubDict ..> client : keys are client*
@enduml

flowchart TB
  subgraph server_maps["server forward indexes"]
    PC["pubsub_channels\nnews -> {A,B}"]
    PP["pubsub_patterns\nnews* -> {C}"]
  end
  subgraph client_maps["client reverse indexes"]
    A["A.pubsub_channels\n{news}"]
    B["B.pubsub_channels\n{news, sports}"]
    C["C.pubsub_patterns\n{news*}"]
  end
  PUB["PUBLISH news"] --> PC
  PUB --> PP
  PC --> A
  PC --> B
  PP --> C

/* pubsub.c — pubsubSubscribeChannel (sketch) */
/* 1) client.pubsub_channels[channel] = present */
/* 2) server.pubsub_channels[channel] → dictAdd(clients, c) */
/* 3) addReplyPubsubSubscribed(c, channel, type) */
markClientAsPubSub(c);

/* Complete PUBLISH path (sharded=0). SPUBLISH: skip patterns; shard bus peers only. */

/* ---- publisher node ---- */

/* 1. Command entry */
void publishCommand(client *c) {
    int receivers = pubsubPublishMessageAndPropagateToCluster(
                        c->argv[1], c->argv[2], /* sharded */ 0);
    /* 11. Standalone / primary→replica (no cluster): replicate the command */
    if (!server.cluster_enabled)
        forceCommandPropagation(c, PROPAGATE_REPL);
    /* 12. Reply: local receivers only */
    addReplyLongLong(c, receivers);
}

/* 2. Local first, then cluster bus */
int pubsubPublishMessageAndPropagateToCluster(robj *channel, robj *message,
                                              int sharded) {
    int receivers = pubsubPublishMessage(channel, message, sharded);
    /* → pubsubPublishMessageInternal(..., pubSubType | pubSubShardType) */
    if (server.cluster_enabled)
        clusterPropagatePublish(channel, message, sharded);  /* step 6+ */
    return receivers;
}

/* 3–5. Local fan-out into subscriber output buffers */
int pubsubPublishMessageInternal(robj *channel, robj *message, pubsubtype type) {
    int receivers = 0;
    unsigned int slot = 0;
    if (server.cluster_enabled && type.shard)
        slot = keyHashSlot(channel->ptr, sdslen(channel->ptr));

    /* 3. Exact: kvstoreDictFind → for each client* addReplyPubsubMessage */
    dictEntry *de = kvstoreDictFind(*type.serverPubSubChannels, slot, channel);
    if (de) { /* iterate dictGetVal(de); receivers++ */ }

    /* 4. SPUBLISH: return here (no patterns) */
    if (type.shard) return receivers;

    /* 5. PUBLISH patterns: for each server.pubsub_patterns entry,
     *    stringmatchlen; addReplyPubsubPatMessage; receivers++ */
    return receivers;
}

/* 6. Build bus message (channel_len + message_len + bulk bytes) */
clusterMsgSendBlock *clusterCreatePublishMsgBlock(robj *channel, robj *message,
                                                  uint16_t type) {
    /* type = CLUSTERMSG_TYPE_PUBLISH or CLUSTERMSG_TYPE_PUBLISHSHARD
     * memcpy channel then message into hdr->data.publish.msg.bulk_data */
    return /* msgblock */;
}

/* 7. Choose who gets the bus packet */
void clusterPropagatePublish(robj *channel, robj *message, int sharded) {
    clusterMsgSendBlock *msgblock;
    if (!sharded) {
        msgblock = clusterCreatePublishMsgBlock(channel, message,
                                                CLUSTERMSG_TYPE_PUBLISH);
        clusterBroadcastMessage(msgblock);           /* step 8 */
        clusterMsgSendBlockDecrRefCount(msgblock);
        return;
    }
    /* SPUBLISH */
    msgblock = clusterCreatePublishMsgBlock(channel, message,
                                            CLUSTERMSG_TYPE_PUBLISHSHARD);
    /* for each node in clusterGetNodesInMyShard (except MYSELF|HANDSHAKE):
     *     clusterSendMessage(node->link, msgblock);     step 9 */
    clusterMsgSendBlockDecrRefCount(msgblock);
}

/* 8. PUBLISH: every known cluster node except self */
void clusterBroadcastMessage(clusterMsgSendBlock *msgblock) {
    dictIterator di;
    dictEntry *de;
    dictInitSafeIterator(&di, server.cluster->nodes);
    while ((de = dictNext(&di)) != NULL) {
        clusterNode *node = dictGetVal(de);
        if (node->flags & (CLUSTER_NODE_MYSELF|CLUSTER_NODE_HANDSHAKE))
            continue;
        clusterSendMessage(node->link, msgblock);    /* step 9 */
    }
    dictResetIterator(&di);
}

/* 9. Write msgblock onto that node’s cluster bus link (async send buffer) */
void clusterSendMessage(clusterLink *link, clusterMsgSendBlock *msgblock);

/* ---- peer node (bus receive) ---- */

/* 10. Handler for CLUSTERMSG_TYPE_PUBLISH / PUBLISHSHARD */
void clusterHandlePublish(/* hdr */) {
    if (!sender) return;
    /* Skip object alloc if this node has zero Pub/Sub subscribers. */
    if ((type == CLUSTERMSG_TYPE_PUBLISH
         && serverPubsubSubscriptionCount() > 0)
     || (type == CLUSTERMSG_TYPE_PUBLISHSHARD
         && serverPubsubShardSubscriptionCount() > 0)) {
        /* rebuild channel, message from hdr->data.publish.msg.bulk_data */
        pubsubPublishMessage(channel, message,
                             type == CLUSTERMSG_TYPE_PUBLISHSHARD);
        /* → steps 3–5 again on THIS node’s indexes only */
    }
}

/* 13. Later on each node: event loop / I/O threads write(2) flush
 *     of subscriber reply buffers (not inside publishCommand). */

/* pubsub.c */
typedef struct pubsubtype {
    int shard;
    dict *(*clientPubSubChannels)(client*);
    int (*subscriptionCount)(client*);
    kvstore **serverPubSubChannels;
    robj **subscribeMsg;
    robj **unsubscribeMsg;
    robj **messageBulk;
} pubsubtype;

pubsubtype pubSubType = { /* SUBSCRIBE / PUBLISH ... */ };
pubsubtype pubSubShardType = { /* SSUBSCRIBE / SPUBLISH ... */ };

Two cluster nodes. Membership is local; nodes do not sync client lists.

  Node N1 (clients A, B)              Node N2 (client C)
  A: SUBSCRIBE news                   C: PSUBSCRIBE news*
  B: SUBSCRIBE news
     SUBSCRIBE sports

N1.pubsub_channels                    N2.pubsub_channels
┌─────────┬──────────┐                (empty for these channels)
│ "news"  │ { A, B } │
│ "sports"│ { B }    │
└─────────┴──────────┘
N1.pubsub_patterns (empty)            N2.pubsub_patterns
                                      ┌─────────┬──────┐
                                      │ "news*" │ { C }│
                                      └─────────┴──────┘
N1 does not know C.  N2 does not know A or B.

Publisher on N1:  PUBLISH news hello

N1 (command target) — pubsubPublishMessage on N1 only
  1) Exact: N1.pubsub_channels["news"] → { A, B }
       addReply* → A, B
  2) Patterns: N1.pubsub_patterns empty → no one
  receivers returned to publisher = 2   (local only; not C)
  3) clusterPropagatePublish → bus carries channel+payload to N2
     (not "A,B are subscribed"; only the message)

N2 (bus receive) — pubsubPublishMessage on N2 only
  1) Exact: N2.pubsub_channels["news"] missing → skip
  2) Patterns: "news*" matches "news" → { C }
       addReply* → C
  N2 never touches A or B

Per-client delivery:  A←N1,  B←N1,  C←N2

Publisher on N2:  PUBLISH sports goal

N2 local: no "sports" channel, no matching pattern → receivers 0
          still propagates payload to N1
N1 local: N1.pubsub_channels["sports"] → { B } → addReply* → B only
          N1 patterns do not match "sports"

UNSUBSCRIBE news from B (still on N1)
  → only N1.pubsub_channels["news"] becomes { A }
  → N2 unchanged

3. Client implementation

Each connected client carries the reverse membership needed for fast unsubscribe, disconnect cleanup, and subscribe-ack counts. The wire replies a subscriber consumes are produced by the server into that client’s output buffer; the client (or library) must parse them and, after faults, re-establish membership.

3.1 Per-client state

/* server.h — client */
dict *pubsub_channels;       /* set of channel robj* this client subscribed to */
dict *pubsub_patterns;       /* set of pattern robj* this client psubscribed to */
dict *pubsubshard_channels;  /* set of shard channel robj* */

Subscription count reported in subscribe acks is dictSize(c->pubsub_channels) + dictSize(c->pubsub_patterns) (plus shard dict for shard commands). The same channel robj (or an equivalent shared key after incrRefCount) appears in both the server map and each member’s client map.

For the two-node example in §2 (A, B on N1; C on N2):

client A (on N1)            client B (on N1)            client C (on N2)
pubsub_channels:            pubsub_channels:            pubsub_channels: (empty)
  { "news" }                  { "news", "sports" }      pubsub_patterns:
pubsub_patterns: (empty)    pubsub_patterns: (empty)      { "news*" }
flags: CLIENT_PUBSUB        flags: CLIENT_PUBSUB        flags: CLIENT_PUBSUB

3.2 Client mode restriction

/* server.c — processCommand */
if ((c->flags & CLIENT_PUBSUB && c->resp == 2) &&
    /* not (P|S)SUBSCRIBE / (P|S)UNSUBSCRIBE / PING / QUIT / RESET */)
{
    rejectCommandFormat(c, "Can't execute '%s': only …");
}

RESP3 connections can use push messages while still issuing other commands more freely; RESP2 Pub/Sub mode is the classic restricted session.

3.3 Message wire shape

RESP3 uses push-style framing (addReplyPushLen) with the same logical fields. These are pushed into the subscriber’s buffer asynchronously relative to the publisher’s command; the subscriber’s client must read them as unsolicited (or push) traffic, not as replies to its own last command (aside from subscribe/unsubscribe acks and PING).

3.4 Reconnect, re-subscribe, and client heartbeats

There is no server-side reconnect or resubscribe. Membership does not survive the connection. After disconnect:

1. Detect socket death (TCP error, library event).
2. Reconnect.
3. Issue SUBSCRIBE / PSUBSCRIBE / SSUBSCRIBE again for the same names.
4. Accept gap loss, or use another primitive (e.g. Streams) if loss is unacceptable.

Libraries such as Redisson (RTopic / RPatternTopic / RShardedTopic) perform reconnect and re-subscribe listeners automatically, but still document that messages published during absence are lost. Redisson’s reliable topic APIs use a different design when gap delivery matters.

PING while subscribed is allowed in RESP2 Pub/Sub mode; the server replies with a Pub/Sub-shaped PONG. That is useful as an application heartbeat because the server idle timeout does not cull CLIENT_PUBSUB clients (see §4.3). Client libraries often ping or rely on connection events and re-subscribe after reconnect.

Server:  no Pub/Sub idle timeout; tcp-keepalive + I/O failure + buffer limits
Client:  must detect socket death; may PING; must re-SUBSCRIBE after reconnect

4. Message delivery and security

Pub/Sub indexes only track who is subscribed. They do not protect messages in transit or after a fault. Delivery is best-effort; access can be restricted with ACL.

4.1 What PUBLISH actually guarantees

PUBLISH
  → for each subscriber: addReply*(output buffer)
  → return receivers (= how many clients were enqueued locally)
  → later: event loop write(2) / I/O thread flush

TCP may retransmit packets while a connection remains up. That is not Redis Pub/Sub retry. Once the client connection is closed, its subscription entries are removed and any unflushed buffer is discarded.

4.2 Temporary network drop

sequenceDiagram
  participant S as Subscriber
  participant R as Redis
  participant P as Publisher
  S->>R: SUBSCRIBE news
  P->>R: PUBLISH news m1
  R-->>S: m1
  Note over S,R: network drop - connection closed
  R->>R: remove S from pubsub_channels
  P->>R: PUBLISH news m2
  Note over R: S not in map - m2 never queued for S
  S->>R: reconnect + SUBSCRIBE news
  P->>R: PUBLISH news m3
  R-->>S: m3

4.3 Availability detection (client and server)

Pub/Sub has no dedicated heartbeat protocol (no server-driven ping of subscribers, no per-message ACK). Liveness is inferred from the TCP connection and from a few Redis/client mechanisms that are not Pub/Sub-specific.

Server side

/* timeout.c — clientsCronHandleTimeout */
if (server.maxidletime &&
    !(c->flags & CLIENT_SLAVE) &&
    !mustObeyClient(c) &&
    !(c->flags & CLIENT_BLOCKED) &&
    !(c->flags & CLIENT_PUBSUB) &&  /* no idle timeout for Pub/Sub */
    (now - c->lastinteraction > server.maxidletime))
{
    freeClient(c);
}

Implication: a half-open TCP session can leave a subscriber in pubsub_channels until TCP keepalive (or the next failed write on PUBLISH) aborts the connection. Until then, PUBLISH may still count that client as a receiver and enqueue into a buffer that will never be read.

Client side

See §3.4: TCP errors, optional PING, and library reconnect/re-subscribe. There is no mutual “are you still interested in channel X?” exchange beyond the existence of the TCP session and optional client PING.

4.4 Transient path loss, TCP recovery, and when Redis frees the client

PUBLISH does not wait for subscriber acknowledgement. It enqueues with addReply* and returns receivers; the socket flush runs asynchronously (handleClientsWithPendingWrites / AE_WRITABLE → writeToClient). The server removes the client from Pub/Sub indexes only when the connection is reclaimed (freeClient / freeClientAsync), typically after a failed write or TCP abort—not during publishCommand itself.

PUBLISH
  → addReply* (subscriber remains in pubsub_channels)
  → return receivers
  → later flush via writeToClient
       → connection still valid → TCP delivers / retransmits as usual
       → connection dead on write → freeClientAsync → indexes cleared

/* networking.c — writeToClient (sketch) */
if (nwritten == -1) {
    if (connGetState(c->conn) != CONN_STATE_CONNECTED) {
        freeClientAsync(c);
        return C_ERR;
    }
}

Role of TCP under a temporary network interruption

Provided both endpoints remain up and the TCP connection control block is retained, TCP is designed to absorb short path disruptions: lost segments are retransmitted while the session stays ESTABLISHED. If the path is restored before the local stack aborts the connection, in-flight and not-yet-acknowledged data—including Pub/Sub payloads already accepted into the send path—can be delivered without Redis tearing down the subscription.

Under common defaults this abort budget is intentionally long relative to brief faults:

Consequently, a temporary network interruption that recovers while client and server processes remain healthy, and while the TCP session has not been aborted, is ordinarily handled by TCP: the subscriber need not re-SUBSCRIBE, and pending traffic can complete. That is transport recovery, not a Redis Pub/Sub replay log—messages never accepted onto a live session are still outside this guarantee.

Same-cloud deployments

When publisher, Redis, and subscribers run inside the same cloud network (for example a single VPC / VNet, private addressing, no consumer-grade NAT on the path), the conditions that most often cut TCP sessions early—aggressive NAT idle timeouts, flaky edge middleboxes—are typically weaker or absent. Retransmission and Redis’s per-socket keepalive then dominate detection latency. Short regional blips that clear within the TCP give-up window are therefore more likely to leave Pub/Sub membership intact and to allow in-flight delivery to finish than paths that traverse the public Internet or carrier-grade NAT.

Same-cloud placement does not create durability: process crash, explicit RST, security-group or LB idle policy, or exhaustion of the retransmission/keepalive budget still ends the session and requires re-subscription, with no replay of the gap.

When Redis frees the client (timing)

Transient path loss, session retained, path recovers
    → TCP retransmit succeeds → subscription kept; delivery can complete

Session aborted (RST, retries/keepalive exhausted, process death)
    → freeClient* → re-SUBSCRIBE required; gap not replayed

Until reclamation, the client may still appear in pubsub_channels and contribute to subsequent receivers counts. After reclamation, further publishes omit that client until it subscribes again.

4.5 Buffer and slow consumers

Even without a full disconnect, a subscriber can lose Pub/Sub traffic if its output buffer hits configured soft/hard limits: Redis may disconnect the client, which again drops membership and pending bytes. Fast publishers plus slow readers are a common failure mode for “I was subscribed but missed messages.”

4.6 Access control (who may pub/sub)

Delivery reliability is separate from authorization. Redis ACL can restrict which channels a user may publish to or subscribe to (resetchannels / channel patterns on the ACL selector; acl-pubsub-default controls the default). Revoking channel permissions can force affected Pub/Sub clients to disconnect. That limits who participates; it does not add persistence or retry.

Payload confidentiality (encryption) is likewise outside Pub/Sub: use TLS on the connection and/or encrypt at the application layer. Channel names and messages are otherwise plaintext RESP on the wire (or inside TLS).

4.7 Choosing a primitive