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

Code before:
    }

    clusterMsg *hdr = (clusterMsg *)link->rcvbuf;
    uint16_t type = ntohs(hdr->type);
    mstime_t now = mstime();

    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;
    }

    /* 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 && (type == CLUSTERMSG_TYPE_PUBLISH_LIGHT || type == CLUSTERMSG_TYPE_PUBLISHSHARD_LIGHT) &&
        nodeSupportsLightMsgHdr(sender)) {
        return pubsubProcessLightPacket(link, type);
    }


    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 && (flags & CLUSTER_NODE_LIGHT_HDR_SUPPORTED)) {
        sender->flags |= CLUSTER_NODE_LIGHT_HDR_SUPPORTED;
    }

    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);
        }
        /* Update the replication offset info for this node. */
        sender->repl_offset = ntohu64(hdr->offset);

Review comment (security):
There's an address sanitizer error for this function, that `hdr->flags` is accessed for a light message.

We have to reorder things in this function. We can't access fields like `hdr->flags` or `hdr->mflags` for light messages.

Suggested order of things:

1. Stuff like sender->data_received = now;
2. If type == light, process light and return.
3. Stuff like `flags = ntohs(hdr->flags)` (needs a non-light message).

Code after:
    mstime_t now = mstime();
    int is_light = IS_LIGHT_MESSAGE(ntohs(hdr->type));
    uint16_t type = ntohs(hdr->type) & ~CLUSTERMSG_MODIFIER_MASK;

    if (is_light) {
        if (!link->node || nodeInHandshake(link->node)) {
            freeClusterLink(link);
            serverLog(
                LL_NOTICE,
                "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;