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

Code before:
    serverAssert(count == 1);

    clusterMsgSendBlock *msgblock, *msgblock_light;

    msgblock_light = clusterCreatePublishLightMsgBlock(
        channel, messages, count, sharded ? CLUSTERMSG_TYPE_PUBLISHSHARD_LIGHT : CLUSTERMSG_TYPE_PUBLISH_LIGHT);

    ClusterNodeIterator iter;
    if (sharded) {
        list *nodes_for_slot = clusterGetNodesInMyShard(server.cluster->myself);
        serverAssert(nodes_for_slot != NULL);
        clusterNodeIterInitMyShard(&iter, nodes_for_slot);
    } else {
        clusterNodeIterInitAllNodes(&iter, server.cluster->nodes);
    }

    clusterNode *node;
    while ((node = clusterNodeIterNext(&iter)) != NULL) {
        if (node->flags & (CLUSTER_NODE_MYSELF | CLUSTER_NODE_HANDSHAKE)) continue;
        /* If the cluster is not marked as homogeneous send respective message blocks to all nodes. */
        if (nodeSupportsLightMsgHdr(node)) {
            clusterSendMessage(node->link, msgblock_light);
        } else {
            msgblock = clusterCreatePublishMsgBlock(channel, messages[0],
                                                    sharded ? CLUSTERMSG_TYPE_PUBLISHSHARD : CLUSTERMSG_TYPE_PUBLISH);
            clusterSendMessage(node->link, msgblock);
            clusterMsgSendBlockDecrRefCount(msgblock);
        }
    }
    clusterNodeIterReset(&iter);
    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. */

Review comment (refactor):
Here we create a new msgblock for every node that needs it. Do you think it will be very rare that more than one node needs it?

I was thinking we initialize it to NULL first, and we create it here only if msgblock == NULL, so we only create it at most once. That's what I meant by lazy initialization. After the loop, we can call clusterMsgSendBlockDecrRefCount only if msgblock != NULL.

Do you think this would be too complex?

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

/* Send a FAILOVER_AUTH_ACK message to the specified node. */
void clusterSendFailoverAuth(clusterNode *node) {
    if (!node->link) return;

    uint32_t msglen = sizeof(clusterMsg) - sizeof(union clusterMsgData);
    clusterMsgSendBlock *msgblock = createClusterMsgSendBlock(CLUSTERMSG_TYPE_FAILOVER_AUTH_ACK, msglen);

    clusterSendMessage(node->link, msgblock);
    clusterMsgSendBlockDecrRefCount(msgblock);
}

/* Send a MFSTART message to the specified node. */
void clusterSendMFStart(clusterNode *node) {
    if (!node->link) return;

    uint32_t msglen = sizeof(clusterMsg) - sizeof(union clusterMsgData);
    clusterMsgSendBlock *msgblock = createClusterMsgSendBlock(CLUSTERMSG_TYPE_MFSTART, msglen);

    clusterSendMessage(node->link, msgblock);
    clusterMsgSendBlockDecrRefCount(msgblock);
}

/* Vote for the node asking for our vote if there are the conditions. */
void clusterSendFailoverAuthIfNeeded(clusterNode *node, clusterMsg *request) {
    clusterNode *primary = node->replicaof;
    uint64_t requestCurrentEpoch = ntohu64(request->currentEpoch);
    uint64_t requestConfigEpoch = ntohu64(request->configEpoch);
    unsigned char *claimed_slots = request->myslots;