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

Code before:
    /* 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 (!node->link || node->pong_received < node->link->ctime) 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
 * -------------------------------------------------------------------------- */

/* 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);
    clusterMsgSendBlock *msgblock = createClusterMsgSendBlock(CLUSTERMSG_TYPE_FAILOVER_AUTH_REQUEST, msglen);

    /* If this is a manual failover, set the CLUSTERMSG_FLAG0_FORCEACK bit
     * in the header to communicate the nodes receiving the message that
     * they should authorized the failover even if the primary is working. */
    if (server.cluster->mf_end) msgblock->data[0].msg.mflags[0] |= CLUSTERMSG_FLAG0_FORCEACK;
    clusterBroadcastMessage(msgblock);
    clusterMsgSendBlockDecrRefCount(msgblock);
}

/* Send a FAILOVER_AUTH_ACK message to the specified node. */
void clusterSendFailoverAuth(clusterNode *node) {

Review comment (bug):
`clusterSendMessage` does an early exit if the link is not established. Isn't that sufficient ? We would have seen crashes by now if it was an issue. WDYT?

Code after:
 * 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
 * -------------------------------------------------------------------------- */

/* 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);
    clusterMsgSendBlock *msgblock = createClusterMsgSendBlock(CLUSTERMSG_TYPE_FAILOVER_AUTH_REQUEST, msglen);