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

Code before:
        senderCurrentEpoch = ntohu64(hdr->currentEpoch);
        senderConfigEpoch = ntohu64(hdr->configEpoch);
        if (senderCurrentEpoch > server.cluster->currentEpoch) server.cluster->currentEpoch = senderCurrentEpoch;
        /* Update the sender configEpoch if it is a primary publishing a newer one. */
        if (!memcmp(hdr->replicaof, CLUSTER_NODE_NULL_NAME, sizeof(hdr->replicaof)) &&
            senderConfigEpoch > sender->configEpoch) {
            sender->configEpoch = senderConfigEpoch;
            clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG | CLUSTER_TODO_FSYNC_CONFIG);
        }
        /* 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);
        }
    }

    /* Initial processing of PING and MEET requests replying with a PONG. */
    if (type == CLUSTERMSG_TYPE_PING || type == CLUSTERMSG_TYPE_MEET) {
        /* We use incoming MEET messages in order to set the address
         * for 'myself', since only other cluster nodes will send us
         * MEET messages on handshakes, when the cluster joins, or
         * later if we changed address, and those nodes will use our
         * official address to connect to us. So by obtaining this address
         * from the socket is a simple way to discover / update our own
         * address in the cluster without it being hardcoded in the config.
         *
         * However if we don't have an address at all, we update the address
         * even with a normal PING packet. If it's wrong it will be fixed
         * by MEET later. */
        if ((type == CLUSTERMSG_TYPE_MEET || myself->ip[0] == '\0') && server.cluster_announce_ip == NULL) {
            char ip[NET_IP_STR_LEN];

            if (connAddrSockName(link->conn, ip, sizeof(ip), NULL) != -1 && strcmp(ip, myself->ip)) {
                memcpy(myself->ip, ip, NET_IP_STR_LEN);
                serverLog(LL_NOTICE, "IP address for this node updated to %s", myself->ip);
                clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG);
            }
        }

        /* Add this node if it is new for us and the msg type is MEET.
         * In this stage we don't try to add the node with the right
         * flags, replicaof pointer, and so forth, as this details will be

Review comment (suggestion):
Missing this?

```suggestion
    } else {
        server.cluster->is_light_hdr_supported = 0;
    }
```

Code after:
            freeClusterLink(link);
            serverLog(LL_WARNING, "Closing link for matching packet type %hu", type);
            return 0;
        }
        return 1;
    }

    clusterMsg *hdr = (clusterMsg *)link->rcvbuf;
    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. */