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

Code before:
 * received from the wrong sender ID). */
int clusterProcessPacket(clusterLink *link) {
    /* Validate that the packet is well-formed */
    if (!clusterIsValidPacket(link)) {
        clusterMsg *hdr = (clusterMsg *)link->rcvbuf;
        uint16_t type = ntohs(hdr->type);
        if (server.debug_cluster_close_link_on_packet_drop &&
            (type == server.cluster_drop_packet_filter || server.cluster_drop_packet_filter == -2)) {
            freeClusterLink(link);
            serverLog(LL_WARNING, "Closing link for matching packet type %s", clusterGetMessageTypeString(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)) {
            serverLog(
                LL_NOTICE,
                "Closing link for node %.40s that sent a lightweight message of type %s as its first message on the link",
                link->node ? link->node->name : "",
                clusterGetMessageTypeString(type));
            freeClusterLink(link);
            return 0;
        }
        clusterNode *sender = link->node;
        sender->data_received = now;
        clusterProcessLightPacket(sender, 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);

    /* We store this information at the link layer so that we can send extensions
     * during the handshake even if we don't know the sender. */
    if (hdr->mflags[0] & CLUSTERMSG_FLAG0_EXT_DATA) {
        link->flags |= CLUSTER_LINK_EXTENSIONS_SUPPORTED;
    }

    /* Store some flags about the sender. */
    if (sender) {

Review comment (suggestion):
Shall we display some placeholder instead of just blank node name?

```suggestion
                link->node ? link->node->name : "(unknown)",
```

Code after:

    return 1;
}

/* When this function is called, there is a packet to process starting
 * at link->rcvbuf. Releasing the buffer is up to the caller, so this
 * function should just handle the higher level stuff of processing the
 * packet, modifying the cluster state if needed.
 *
 * The function returns 1 if the link is still valid after the packet
 * was processed, otherwise 0 if the link was freed since the packet
 * processing lead to some inconsistency error (for instance a PONG
 * received from the wrong sender ID). */
int clusterProcessPacket(clusterLink *link) {
    /* Validate that the packet is well-formed */
    if (!clusterIsValidPacket(link)) {
        clusterMsg *hdr = (clusterMsg *)link->rcvbuf;
        uint16_t type = ntohs(hdr->type);
        if (server.debug_cluster_close_link_on_packet_drop &&
            (type == server.cluster_drop_packet_filter || server.cluster_drop_packet_filter == -2)) {
            freeClusterLink(link);
            serverLog(LL_WARNING, "Closing link for matching packet type %s", clusterGetMessageTypeString(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)) {
            serverLog(
                LL_NOTICE,
                "Closing link for node %.40s (%s) that sent a lightweight message of type %s as its first message on the link",
                clusterLinkGetNodeName(link),
                clusterLinkGetHumanNodeName(link),
                clusterGetMessageTypeString(type));
            freeClusterLink(link);
            return 0;
        }
        clusterNode *sender = link->node;
        sender->data_received = now;
        clusterProcessLightPacket(sender, link, type);
        return 1;
    }

    uint16_t flags = ntohs(hdr->flags);
    uint64_t sender_claimed_current_epoch = 0, sender_claimed_config_epoch = 0;