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

Code before:
     * Pub/Sub subscribers. */
    if ((type == CLUSTERMSG_TYPE_PUBLISH && serverPubsubSubscriptionCount() > 0) ||
        (type == CLUSTERMSG_TYPE_PUBLISHSHARD && serverPubsubShardSubscriptionCount() > 0)) {
        channel_len = ntohl(publish_data->channel_len);
        message_len = ntohl(publish_data->message_len);
        channel = createStringObject((char *)publish_data->bulk_data, channel_len);
        message = createStringObject((char *)publish_data->bulk_data + channel_len, message_len);
        pubsubPublishMessage(channel, message, type == CLUSTERMSG_TYPE_PUBLISHSHARD);
        decrRefCount(channel);
        decrRefCount(message);
    }
}

static void clusterProcessModulePacket(clusterMsgModule *module_data, clusterNode *sender) {
    if (!sender) return; /* Protect the module from unknown nodes. */
    /* We need to route this message back to the right module subscribed
     * for the right message type. */
    uint64_t module_id = module_data->module_id; /* Endian-safe ID */
    uint32_t len = ntohl(module_data->len);
    uint8_t type = module_data->type;
    unsigned char *payload = module_data->bulk_data;

    /* Ensure sender name is properly truncated to 40 characters */
    sds truncated_name = sdsnewlen(sender->name, 40);
    moduleCallClusterReceivers(truncated_name, module_id, type, payload, len);
    sdsfree(truncated_name);
}

static void clusterProcessLightPacket(clusterNode *sender, clusterLink *link, uint16_t type) {
    clusterMsgLight *hdr = (clusterMsgLight *)link->rcvbuf;
    serverLog(LL_DEBUG, "Processing light packet of type: %s", clusterGetMessageTypeString(type));
    if (type == CLUSTERMSG_TYPE_PUBLISH || type == CLUSTERMSG_TYPE_PUBLISHSHARD) {
        clusterProcessPublishPacket(&hdr->data.publish.msg, type);
    } else if (type == CLUSTERMSG_TYPE_MODULE) {
        clusterProcessModulePacket(&hdr->data.module.msg, sender);
    } else {
        serverAssert(0);
    }
}

static inline int messageTypeSupportsLightHdr(uint16_t type) {
    switch (type) {
    case CLUSTERMSG_TYPE_PUBLISH: return 1;
    case CLUSTERMSG_TYPE_PUBLISHSHARD: return 1;
    case CLUSTERMSG_TYPE_MODULE: return 1;
    }
    return 0;
}


int clusterIsValidPacket(clusterLink *link) {

Review comment (suggestion):
This is not truncating per se. This is more like adding a `\0` at the end (null terminated string). You could just name it as sender_id.

```suggestion
    sds sender_name = sdsnewlen(sender->name, 40);
    moduleCallClusterReceivers(sender_name, module_id, type, payload, len);
    sdsfree(sender_name);
```    


Code after:
     * Pub/Sub subscribers. */
    if ((type == CLUSTERMSG_TYPE_PUBLISH && serverPubsubSubscriptionCount() > 0) ||
        (type == CLUSTERMSG_TYPE_PUBLISHSHARD && serverPubsubShardSubscriptionCount() > 0)) {
        channel_len = ntohl(publish_data->channel_len);
        message_len = ntohl(publish_data->message_len);
        channel = createStringObject((char *)publish_data->bulk_data, channel_len);
        message = createStringObject((char *)publish_data->bulk_data + channel_len, message_len);
        pubsubPublishMessage(channel, message, type == CLUSTERMSG_TYPE_PUBLISHSHARD);
        decrRefCount(channel);
        decrRefCount(message);
    }
}

static void clusterProcessModulePacket(clusterMsgModule *module_data, clusterNode *sender) {
    if (!sender) return; /* Protect the module from unknown nodes. */
    /* We need to route this message back to the right module subscribed
     * for the right message type. */
    uint64_t module_id = module_data->module_id; /* Endian-safe ID */
    uint32_t len = ntohl(module_data->len);
    uint8_t type = module_data->type;
    unsigned char *payload = module_data->bulk_data;

    sds sender_name = sdsnewlen(sender->name, CLUSTER_NAMELEN);
    moduleCallClusterReceivers(sender_name, module_id, type, payload, len);
    sdsfree(sender_name);
}

static void clusterProcessLightPacket(clusterNode *sender, clusterLink *link, uint16_t type) {
    clusterMsgLight *hdr = (clusterMsgLight *)link->rcvbuf;
    serverLog(LL_DEBUG, "Processing light packet of type: %s", clusterGetMessageTypeString(type));
    if (type == CLUSTERMSG_TYPE_PUBLISH || type == CLUSTERMSG_TYPE_PUBLISHSHARD) {
        clusterProcessPublishPacket(&hdr->data.publish.msg, type);
    } else if (type == CLUSTERMSG_TYPE_MODULE) {
        clusterProcessModulePacket(&hdr->data.module.msg, sender);
    } else {
        serverAssert(0);
    }
}

static inline int messageTypeSupportsLightHdr(uint16_t type) {
    switch (type) {
    case CLUSTERMSG_TYPE_PUBLISH: return 1;
    case CLUSTERMSG_TYPE_PUBLISHSHARD: return 1;
    case CLUSTERMSG_TYPE_MODULE: return 1;
    }
    return 0;
}


int clusterIsValidPacket(clusterLink *link) {
    clusterMsg *hdr = (clusterMsg *)link->rcvbuf;