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

Code before:

/* -----------------------------------------------------------------------------
 * 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 **messages, int count, int sharded) {
    clusterMsgSendBlock *msgblock, *msgblock_light;
    int i;
    msgblock_light = clusterCreatePublishLightMsgBlock(
        channel, messages, count, sharded ? CLUSTERMSG_TYPE_PUBLISHSHARD_LIGHT : CLUSTERMSG_TYPE_PUBLISH_LIGHT);
    if (!sharded) {
        clusterBroadcastMessage(msgblock_light);
        clusterMsgSendBlockDecrRefCount(msgblock_light);
        if (!server.cluster->is_light_hdr_supported) {
            for (i = 0; i < count; i++) {
                msgblock = clusterCreatePublishMsgBlock(channel, messages[i], CLUSTERMSG_TYPE_PUBLISH);
                clusterBroadcastMessage(msgblock);
                clusterMsgSendBlockDecrRefCount(msgblock);
            }
        }
        return;
    }

    listIter li;
    listNode *ln;
    clusterNode *node;
    list *nodes_for_slot = clusterGetNodesInMyShard(server.cluster->myself);
    serverAssert(nodes_for_slot != NULL);
    listRewind(nodes_for_slot, &li);
    int nodes_not_supporting_light_header = 0;
    while ((ln = listNext(&li))) {
        node = listNodeValue(ln);
        if (node->flags & (CLUSTER_NODE_MYSELF | CLUSTER_NODE_HANDSHAKE)) continue;
        if (nodeSupportsLightMsgHdr(node)) {
            clusterSendMessage(node->link, msgblock_light);
        } else {
            nodes_not_supporting_light_header++;
            continue;
        }
    }
    clusterMsgSendBlockDecrRefCount(msgblock_light);

    if (nodes_not_supporting_light_header) {
        server.cluster->is_light_hdr_supported = 0;
    } else {

Review comment (refactor):
As mentioned in another comment, I think we should ignore `clusterBroadcastMessage` here and just loop over the nodes. We can keep the clusterBroadcastMessage function simple and I think it will be simpler even here. We already do a loop over nodes for sharded pubsub so we can do the same for unsharded and sharded, just loop over another list of nodes.

I wanted to see how complex it gets, so I wrote the (untested!) code below. With this idea, we don't need to store `server.cluster->is_light_hdr_supported` at all. It looks simpler to me, but I probably missed something...

(...) I just realized that the shard nodes is a list, but all nodes is a dict, so the iteration is different. I did a quickfix to the iteration stuff, which looks a bit ugly. :) It avoids duplicated code though. WDYT?

```C
void clusterPropagatePublish(robj *channel, robj **messages, int count, int sharded) {
    /* Lazy create msg blocks later when we need them. */
    clusterMsgSendBlock *msgblock_light = NULL;     /* Just one, multi-message */
    clusterMsgSendBlock *msgblocks[count] = {NULL}; /* Heavy, one per message */
    dictIterator di;                                /* For unsharded */
    listIter li;                                    /* For sharded */
    if (sharded) {
        list *nodes_for_slot = clusterGetNodesInMyShard(server.cluster->myself);
        serverAssert(nodes_for_slot != NULL);
        listRewind(nodes_for_slot, &li);
    } else {
        dictInitSafeIterator(&di, server.cluster->nodes);
    }
    do {
        clusterNode *node;
        if (sharded) {
            listNode *ln = listNext(&li);
            if (!ln) break;
            node = listNodeValue(ln);
        } else {
            dictEntry *de = dictNext(&di);
            if (!de) break;
            node = dictGetVal(de);
        }
        if (node->flags & (CLUSTER_NODE_MYSELF | CLUSTER_NODE_HANDSHAKE)) continue;
        if (nodeSupportsLightMsgHdr(node)) {
            if (msgblock_light == NULL) {
                /* Lazy creation */
                msgblock_light = clusterCreatePublishLightMsgBlock(...);
            }
            clusterSendMessage(node->link, msgblock_light);
        } else {
            /* Heavy msgblock, one for each pubsub message. */
            for (i = 0; i < count; i++) {
                if (msgblocks[i] == NULL) {
                    /* Lazy create msgblock */
                    msgblocks[i] = clusterCreatePublishMsgBlock(...);
                }
                clusterSendMessage(node->link, msgblocks[i]);
            }
        }
    }

    /* Cleanup what we created. */
    if (!sharded) dictResetIterator(&di);
    if (light_msgblock != NULL) clusterMsgSendBlockDecrRefCount(light_msgblock);
    if (msgblocks[0] != NULL) {
        for (i = 0; i < count; i++) {
            clusterMsgSendBlockDecrRefCount(msgblocks[i]);
        }
    }
}
```

Code after:
 * -------------------------------------------------------------------------- */
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 (nodeSupportsLightMsgHdr(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);

    clusterMsg *hdr = &msgblock->msg;
    /* 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) hdr->mflags[0] |= CLUSTERMSG_FLAG0_FORCEACK;
    clusterBroadcastMessage(msgblock);
    clusterMsgSendBlockDecrRefCount(msgblock);