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

Code before:
    if (sender) {
        if (flags & CLUSTER_NODE_LIGHT_HDR_SUPPORTED) {
            sender->flags |= CLUSTER_NODE_LIGHT_HDR_SUPPORTED;
        } else {
            sender->flags &= ~CLUSTER_NODE_LIGHT_HDR_SUPPORTED;
        }
    }

    /* Update the last time we saw any data from this node. We
     * use this in order to avoid detecting a timeout from a node that
     * is just sending a lot of data in the cluster bus, for instance
     * because of Pub/Sub. */
    if (sender) sender->data_received = now;

    if (sender && !nodeInHandshake(sender)) {
        /* Update our currentEpoch if we see a newer epoch in the cluster. */
        sender_claimed_current_epoch = ntohu64(hdr->currentEpoch);
        sender_claimed_config_epoch = ntohu64(hdr->configEpoch);
        if (sender_claimed_current_epoch > server.cluster->currentEpoch)
            server.cluster->currentEpoch = sender_claimed_current_epoch;
        /* Update the sender configEpoch if it is a primary publishing a newer one. */
        if (sender_claims_to_be_primary && sender_claimed_config_epoch > sender->configEpoch) {
            sender->configEpoch = sender_claimed_config_epoch;
            clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG | CLUSTER_TODO_FSYNC_CONFIG);

            if (server.cluster->failover_auth_time && sender->configEpoch == server.cluster->failover_auth_epoch) {
                /* There are another node has claimed it in this epoch, if we have any ongoing
                 * election, we can reset it since there won't be enough votes and we can start
                 * a new one ASAP. */
                server.cluster->failover_auth_time = 0;
                serverLog(LL_WARNING,
                          "I have a failover election for epoch %llu in progress and "
                          "received node %.40s (%s) claiming this epoch, resetting the election.",
                          (unsigned long long)sender->configEpoch, sender->name, sender->human_nodename);
                clusterDoBeforeSleep(CLUSTER_TODO_HANDLE_FAILOVER);
            }
        }
        /* Update the replication offset info for this node. */
        sender->repl_offset = ntohu64(hdr->offset);
        sender->repl_offset_time = now;
        /* If we are a replica performing a manual failover and our primary
         * sent its offset while already paused, populate the MF state. */
        if (server.cluster->mf_end && nodeIsReplica(myself) && myself->replicaof == sender &&
            hdr->mflags[0] & CLUSTERMSG_FLAG0_PAUSED && server.cluster->mf_primary_offset == -1) {
            server.cluster->mf_primary_offset = sender->repl_offset;
            clusterDoBeforeSleep(CLUSTER_TODO_HANDLE_MANUALFAILOVER);
            serverLog(LL_NOTICE,
                      "Received replication offset for paused "
                      "primary manual failover: %lld",
                      server.cluster->mf_primary_offset);
        }

Review comment (bug):
```suggestion
            if (server.cluster->failover_auth_time && sender->configEpoch >= server.cluster->failover_auth_epoch) {
```

Code after:
                "Closing link for node that sent a lightweight message of type %hu as its first message on the link",
                type);
            return 0;
        }
        clusterNode *sender = link->node;
        sender->data_received = now;
        clusterProcessLightPacket(link, type);
        return 1;
    }

    uint16_t flags = ntohs(hdr->flags);
    uint64_t sender_claimed_current_epoch = 0, sender_claimed_config_epoch = 0;
    clusterNode *sender = getNodeFromLinkAndMsg(link, hdr);
    int sender_claims_to_be_primary = !memcmp(hdr->replicaof, CLUSTER_NODE_NULL_NAME, CLUSTER_NAMELEN);
    int sender_last_reported_as_replica = sender && nodeIsReplica(sender);
    int sender_last_reported_as_primary = sender && nodeIsPrimary(sender);

    if (sender && (hdr->mflags[0] & CLUSTERMSG_FLAG0_EXT_DATA)) {
        sender->flags |= CLUSTER_NODE_EXTENSIONS_SUPPORTED;
    }

    /* Checks if the node supports light message hdr */
    if (sender) {
        if (flags & CLUSTER_NODE_LIGHT_HDR_SUPPORTED) {
            sender->flags |= CLUSTER_NODE_LIGHT_HDR_SUPPORTED;
        } else {
            sender->flags &= ~CLUSTER_NODE_LIGHT_HDR_SUPPORTED;
        }
    }

    /* Update the last time we saw any data from this node. We
     * use this in order to avoid detecting a timeout from a node that
     * is just sending a lot of data in the cluster bus, for instance
     * because of Pub/Sub. */
    if (sender) sender->data_received = now;

    if (sender && !nodeInHandshake(sender)) {
        /* Update our currentEpoch if we see a newer epoch in the cluster. */
        sender_claimed_current_epoch = ntohu64(hdr->currentEpoch);
        sender_claimed_config_epoch = ntohu64(hdr->configEpoch);
        if (sender_claimed_current_epoch > server.cluster->currentEpoch)
            server.cluster->currentEpoch = sender_claimed_current_epoch;
        /* Update the sender configEpoch if it is a primary publishing a newer one. */
        if (sender_claims_to_be_primary && sender_claimed_config_epoch > sender->configEpoch) {
            sender->configEpoch = sender_claimed_config_epoch;
            clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG | CLUSTER_TODO_FSYNC_CONFIG);

            if (server.cluster->failover_auth_time && sender->configEpoch >= server.cluster->failover_auth_epoch) {
                /* Another node has claimed an epoch greater than or equal to ours.
                 * If we have an ongoing election, reset it because we cannot win
                 * with an epoch smaller than or equal to the incoming claim. This