Repository: valkey-io/valkey
File: src/cluster_legacy.c (C)

Code before:
 *
 * If `sharded` is 0:
 * For now we do very little, just propagating [S]PUBLISH messages across the whole
 * cluster. In the future we'll try to get smarter and avoiding propagating those
 * messages to hosts without receives for a given channel.
 * Otherwise:
 * Publish this message across the slot (primary/replica).
 * -------------------------------------------------------------------------- */
void clusterPropagatePublish(robj *channel, robj *message, int sharded) {
    clusterMsgSendBlock *msgblock, *msgblock_light;
    msgblock_light = clusterCreatePublishMsgBlock(channel, message, 1, sharded);
    /* We will only create msgblock with normal hdr if there are any nodes that do not support light hdr */
    msgblock = NULL;
    ClusterNodeIterator iter;
    if (sharded) {
        clusterNodeIterInitMyShard(&iter);
    } else {
        clusterNodeIterInitAllNodes(&iter);
    }

    clusterNode *node;
    while ((node = clusterNodeIterNext(&iter)) != NULL) {
        if (node->flags & (CLUSTER_NODE_MYSELF | CLUSTER_NODE_HANDSHAKE)) continue;
        if (nodeSupportsLightMsgHdrForPubSub(node)) {
            /* Avoid sending pub/sub light weight message until the bidirectional link(s) have been established. */
            if (!node->link || node->pong_received < node->link->ctime) continue;
            clusterSendMessage(node->link, msgblock_light);
        } else {
            if (msgblock == NULL) {
                msgblock = clusterCreatePublishMsgBlock(channel, message, 0, sharded);
            }
            clusterSendMessage(node->link, msgblock);
        }
    }
    clusterNodeIterReset(&iter);
    if (msgblock != NULL) clusterMsgSendBlockDecrRefCount(msgblock);
    clusterMsgSendBlockDecrRefCount(msgblock_light);
}

/* -----------------------------------------------------------------------------
 * REPLICA node specific functions
 * -------------------------------------------------------------------------- */

/* This function sends a FAILOVER_AUTH_REQUEST message to every node in order to
 * see if there is the quorum for this replica instance to failover its failing
 * primary.
 *
 * Note that we send the failover request to everybody, primary and replica nodes,
 * but only the primaries are supposed to reply to our query. */
void clusterRequestFailoverAuth(void) {
    uint32_t msglen = sizeof(clusterMsg) - sizeof(union clusterMsgData);

Review comment (bug):
Good finding!

The bug was introduced when we added the light message, so I'm thinking we should fall back to the heavy-weight message instead of just skipping the message propagation.

```suggestion
        if (nodeSupportsLightMsgHdrForPubSub(node) && node->link && node->pong_received >= node->link->ctime) {
            /* Avoid sending pub/sub light weight message until the bidirectional link(s) have been established. */
```

Maybe even better is to do the check for `node->pong_received >= node->link->ctime` inside the definition of `nodeSupportsLightMsgHdrForPubSub()`?

Code after:

    if (target != NULL) {
        node = clusterLookupNode(target, strlen(target));
        if (node == NULL || node->link == NULL) return C_ERR;
    }

    clusterSendModule(target ? node->link : NULL, module_id, type, payload, len);
    return C_OK;
}

/* -----------------------------------------------------------------------------
 * CLUSTER Pub/Sub support
 *
 * If `sharded` is 0:
 * For now we do very little, just propagating [S]PUBLISH messages across the whole
 * cluster. In the future we'll try to get smarter and avoiding propagating those
 * messages to hosts without receives for a given channel.
 * Otherwise:
 * Publish this message across the slot (primary/replica).
 * -------------------------------------------------------------------------- */
void clusterPropagatePublish(robj *channel, robj *message, int sharded) {
    clusterMsgSendBlock *msgblock, *msgblock_light;
    msgblock_light = clusterCreatePublishMsgBlock(channel, message, 1, sharded);
    /* We will only create msgblock with normal hdr if there are any nodes that do not support light hdr */
    msgblock = NULL;
    ClusterNodeIterator iter;
    if (sharded) {
        clusterNodeIterInitMyShard(&iter);
    } else {
        clusterNodeIterInitAllNodes(&iter);
    }

    clusterNode *node;
    while ((node = clusterNodeIterNext(&iter)) != NULL) {
        if (node->flags & (CLUSTER_NODE_MYSELF | CLUSTER_NODE_HANDSHAKE)) continue;
        if (nodeSupportsLightMsgHdrForPubSub(node)) {
            clusterSendMessage(node->link, msgblock_light);
        } else {
            if (msgblock == NULL) {
                msgblock = clusterCreatePublishMsgBlock(channel, message, 0, sharded);
            }
            clusterSendMessage(node->link, msgblock);
        }
    }
    clusterNodeIterReset(&iter);
    if (msgblock != NULL) clusterMsgSendBlockDecrRefCount(msgblock);
    clusterMsgSendBlockDecrRefCount(msgblock_light);
}

/* -----------------------------------------------------------------------------
 * REPLICA node specific functions