RESP3 push frame torn apart when a pubsub message >= 16384 bytes is delivered to the publishing connection (9.0 regression)

First issue binbin assigned directly to me — I used it to dig into this code path.

Original issue: https://github.com/valkey-io/valkey/issues/4231

1. Business scenario

#

One TCP connection does two things:

1. SUBSCRIBE wiretest — becomes a channel subscriber
2. PUBLISH wiretest <large payload> — is also the publisher

Valkey delivers a copy to every subscriber, including the publisher itself.
SignalR Redis backplane and some multiplexer clients use “same connection pub+sub”, so they hit this.

A connection that only subscribes (never publishes) gets correct delivery. Only the publisher’s own copy is broken.

2. Protocol layer: what the client sees on the wire

#

Valkey speaks RESP. Bytes from read() are a stream of type tags + payloads that must be parsed in order.

RESP2 pubsub delivery is a normal array:

*3\r\n
$7\r\nmessage\r\n
$8\r\nwiretest\r\n
$16384\r\nXXXX...\r\n

RESP3 uses push (>) for “server-initiated push, not a reply to a command”:

>3\r\n
$7\r\nmessage\r\n
$8\r\nwiretest\r\n
$16384\r\nXXXX...\r\n

On the same connection, PUBLISH itself also has an integer reply (receiver count):

:1\r\n

Clients (e.g. StackExchange.Redis) track “what reply this command expects”. They want:

:1\r\n                          ← PUBLISH 的回复
>3\r\n$message\r\n$channel\r\n$payload\r\n   ← 推送

If they first see a bare $16384\r\n..., they treat it as the PUBLISH reply → error and desync with the rest of the stream.

3. Server side: per-connection outbound queue

#

Each client has roughly three write-related pieces (see client in src/server.h):

Write order (write / writev): roughly c->buf first, then c->reply in list order.
Whoever entered buf/reply first goes on the socket first.

Typical event-loop path:

read(query) → 解析命令 → 执行 → 把回复塞进 c->buf / c->reply
           → 事件循环稍后 writev(fd, iov...) 把缓冲刷到内核

4. Pub/Sub delivery: from PUBLISH to bytes

#

Command entry in src/pubsub.c:

void publishCommand(client *c) {
    if (server.sentinel_mode) {
        sentinelPublishCommand(c);
        return;
    }

    int receivers = pubsubPublishMessageAndPropagateToCluster(c->argv[1], c->argv[2], 0);
    if (!server.cluster_enabled) forceCommandPropagation(c, PROPAGATE_REPL);
    addReplyLongLong(c, receivers);
}

Order matters:

1. First pubsubPublishMessage... — enqueue push for all subscribers
2. Then addReplyLongLong(c, receivers) — write :N to the current client

Delivery core:

int pubsubPublishMessageInternal(robj *channel, robj *message, pubsubtype type) {
    ...
    while (hashtableNext(&iter, &c)) {
        addReplyPubsubMessage(c, channel, message, *type.messageBulk);
        ...
        receivers++;
    }

Assemble one push frame:

void addReplyPubsubMessage(client *c, robj *channel, robj *msg, robj *message_bulk) {
    struct ClientFlags old_flags = c->flag;
    c->flag.pushing = 1;
    if (c->resp == 2)
        addReply(c, shared.mbulkhdr[3]);
    else
        addReplyPushLen(c, 3);
    addReply(c, message_bulk);
    addReplyBulk(c, channel);
    if (msg) addReplyBulk(c, msg);
    if (!old_flags.pushing) c->flag.pushing = 0;
}

Semantically four consecutive writes:

1. >3\r\n (or RESP2 *3)
2. message
3. channel name (bulk)
4. payload (bulk)

c->flag.pushing = 1 tells the reply API: “this is a push, not an ordinary command reply”.

5. Why pending_push_messages (defer)

#

If the publisher is the current connection and the push is written into c->buf mid-PUBLISH, you get:

[push 半帧或整帧] [ :1 ] [或许更多]

Worse with MULTI/EXEC: pushes interleaved among command replies break client state machines.

So Valkey’s rule:

When delivering a push to the client currently executing a command, do not put it on the formal reply queue yet — park it in pending_push_messages; after the command finishes and its reply is written, listJoin it onto the end of c->reply.

Logic in _addReplyToBufferOrList in src/networking.c:

    /* If we're processing a push message into the current client (i.e. executing PUBLISH
     * to a channel which we are subscribed to, then we wanna postpone that message ... */
    int defer_push_message = c->flag.pushing && c == server.current_client && server.executing_client &&
                             !cmdHasPushAsReply(server.executing_client->cmd);
    ...
    if (defer_push_message) {
        _addReplyProtoToList(c, server.pending_push_messages, s, len);
        return;
    }

Conditions:

• c->flag.pushing: assembling a push (addReplyPubsubMessage set this)
• c == server.current_client: push recipient is the command issuer (self-publish)
• Current command is not a SUBSCRIBE-family command (those use push as their “reply”)

After the command:

    if (!server.execution_nesting) listJoin(c->reply, server.pending_push_messages);

Ideal timeline:

执行 PUBLISH
  ├─ 给自己的 push 各段 → pending_push_messages
  └─ :1 → c->buf / c->reply
afterCommand
  └─ pending 接到 c->reply 尾部

写出顺序:  :1  →  完整 >3 push

Delivery to others has c != current_client, so no defer — the whole frame goes straight into their buf/reply and order is naturally correct.

6. Reply Copy Avoidance: the 9.0 optimization

#

Memcpy’ing large bulk strings into the reply buffer wastes CPU/memory bandwidth.
#2078 added “copy avoidance”:

• Reply buffer stores a small bulkStrRef { robj *obj; char *str; }, not the string body
• On real writev, iovecs point into the object’s sds memory

Expand on write (see src/networking.c):

static void addBulkStringToReplyIOV(char *buf, size_t buf_len, replyIOV *reply, bufWriteMetadata *metadata) {
    bulkStrRef *str_ref = (bulkStrRef *)buf;
    while (buf_len > 0 && !reply->limit_reached) {
        size_t str_len = sdslen(str_ref->str);
        /* RESP encodes bulk strings as $<length>\r\n<data>\r\n */
        ...
        addPlainBufferToReplyIOV(... prefix "$len\r\n" ...);
        addPlainBufferToReplyIOV(str_ref->str, str_len, ...);  /* 直接指对象内存 */
        addPlainBufferToReplyIOV(reply->crlf, 2, ...);

Entry in addReplyBulk:

void addReplyBulk(client *c, robj *obj) {
    if (tryAvoidBulkStrCopyToReply(c, obj) == C_OK) {
        ...
        return;   /* 走了 copy-avoid，不再走下面的普通路径 */
    }
    addReplyBulkLen(c, obj);
    addReply(c, obj);           /* → _addReplyToBufferOrList（会 defer） */
    addReplyProto(c, "\r\n", 2);
}

Gated by isCopyAvoidPreferred; single-thread default threshold 16384:

        return server.min_string_size_copy_avoid && sdslen(objectGetVal(obj)) >= (size_t)server.min_string_size_copy_avoid;

So:

• payload 16383: no CA → addReply → _addReplyToBufferOrList → defers → correct
• payload ≥16384: CA → _addBulkStrRefToBufferOrList → ignores defer → bug

Looks like a “16KB chunk” issue; root cause is CA’s default threshold happens to be 16KB.

7. Both paths together: step through the bug

#

Scenario: same connection HELLO 3 + SUBSCRIBE + PUBLISH 16384 bytes.

Step A: start assembling self push

#

addReplyPubsubMessage sets pushing=1, then:

At this point pending_push_messages is roughly:

>3\r\n $7\r\nmessage\r\n $8\r\nwiretest\r\n

Missing the final payload.

Step B: large payload takes the wrong exit

#

tryAvoidBulkStrCopyToReply → _addBulkStrRefToBufferOrList:

static void _addBulkStrRefToBufferOrList(client *c, robj *obj) {
    ...
    /* 没有 defer_push_message 判断！ */
    if (!_addBulkStrRefToBuffer(c, ...)) {
        _addBulkStrRefToToList(c, ...);  /* 硬编码进 c->reply */
    }
}

bulkStrRef lands in c->buf (or c->reply), not pending.

Memory view:

c->buf:     [payloadHeader][bulkStrRef{ptr→argv[2]}]
pending:    >3 ... message ... channel     （缺 payload）

Step C: PUBLISH’s own reply

#

addReplyLongLong(c, 1) → :1\r\n also into c->buf, right after the bulkStrRef:

c->buf:     [bulkStrRef] [:1\r\n]
pending:    >3 ... message ... channel

Step D: afterCommand join

#

listJoin(c->reply, pending):

写出顺序:
  1) 展开 bulkStrRef → $16384\r\nXXXX...\r\n
  2) :1\r\n
  3) >3\r\n$message\r\n$channel\r\n     ← 没有 payload 的残缺 push

That is the issue’s wire capture. Client thinks $16384... is the PUBLISH reply → protocol desync.

8. One diagram to remember

#

                    addReplyPubsubMessage (pushing=1)
                              │
          ┌───────────────────┼───────────────────┐
          │                   │                   │
     小片段 (header/          channel             payload ≥16KB
     message)                 (小 bulk)           │
          │                   │                   │
          ▼                   ▼                   ▼
   _addReplyToBufferOrList              addReplyBulk
          │                                   │
          │ defer?                            ├─ <阈值: 普通路径 → defer ✓
          │ (self+pushing)                    └─ ≥阈值: copy-avoid
          ▼                                         │
   pending_push_messages                            ▼
                                              _addBulkStrRef*
                                              → c->buf / c->reply  ✗
                                                    │
   publishCommand 再写 :1 ──────────────────────────┘
                                                    │
   afterCommand: listJoin(pending → reply 尾)       │
                                                    ▼
                              撕碎的 socket 字节流

9. Suggested reading order (guided tour)

#

1. publishCommand / addReplyPubsubMessage — src/pubsub.c
2. _addReplyToBufferOrList defer comments — src/networking.c ~744
3. afterCommand listJoin — src/server.c ~4195
4. addReplyBulk / isCopyAvoidPreferred / _addBulkStrRefToBufferOrList — same file
5. addBulkStringToReplyIOV — how CA becomes $len\r\ndata

Local check: run an instance with the issue’s Python; or CONFIG GET min-string-size-avoid-copy-reply for the threshold. Setting the threshold to 0 disables CA and large messages work again — side evidence that the root cause is CA bypassing defer, not pubsub itself.