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

Code before:
    ssize_t nread;
    clusterMsg *hdr;
    clusterLink *link = connGetPrivateData(conn);
    unsigned int readlen, rcvbuflen;

    while (1) { /* Read as long as there is data to read. */
        rcvbuflen = link->rcvbuf_len;
        if (rcvbuflen < RCVBUF_MIN_READ_LEN) {
            /* First, obtain the first 16 bytes to get the full message
             * length and type. */
            readlen = RCVBUF_MIN_READ_LEN - rcvbuflen;
        } else {
            /* Finally read the full message. */
            hdr = (clusterMsg *)link->rcvbuf;
            uint16_t type = ntohs(hdr->type);
            if (rcvbuflen == RCVBUF_MIN_READ_LEN) {
                int is_msg_valid = 0;
                /* Perform some sanity check on the message signature
                 * and length. */
                if (memcmp(hdr->sig, "RCmb", 4) != 0 || ntohl(hdr->totlen) < CLUSTERMSG_MIN_LEN) {
                    /* The minimum length for PUBLISH and  PUBLISHSHARD will be clusterMsgLight_MIN_LEN
                     * as we are using the `clusterMsgLight` hdr. */
                    if ((type == CLUSTERMSG_TYPE_PUBLISH_LIGHT || type == CLUSTERMSG_TYPE_PUBLISHSHARD_LIGHT) &&
                        ((ntohl(hdr->totlen) >= CLUSTERMSG_LIGHT_MIN_LEN))) {
                        is_msg_valid = 1;
                    }
                    if (!is_msg_valid) {
                        char ip[NET_IP_STR_LEN];
                        int port;
                        if (connAddrPeerName(conn, ip, sizeof(ip), &port) == -1) {
                            serverLog(LL_WARNING, "Bad message length or signature received "
                                                  "on the Cluster bus.");
                        } else {
                            serverLog(LL_WARNING,
                                      "Bad message length or signature received "
                                      "on the Cluster bus from %s:%d",
                                      ip, port);
                        }
                        handleLinkIOError(link);
                        return;
                    }
                }
            }
            readlen = ntohl(hdr->totlen) - rcvbuflen;
            if (readlen > sizeof(buf)) readlen = sizeof(buf);
        }

        nread = connRead(conn, buf, readlen);
        if (nread == -1 && (connGetState(conn) == CONN_STATE_CONNECTED)) return; /* No more data ready. */

        if (nread <= 0) {

Review comment (refactor):
This is still inside the `if` so a packet can be marked as valid here even if `sig != "RCmb"`. I think we need to refactor this `if` a little.

```C
if (memcmp(hdr->sig, "RCmb", 4) != 0 || ntohl(hdr->totlen) < CLUSTERMSG_MIN_LEN) {
```

Code after:
        moduleCallClusterReceivers(sender->name, module_id, type, payload, len);
    } else {
        serverLog(LL_WARNING, "Received unknown packet type: %d", type);
    }
    return 1;
}

/* This function is called when we detect the link with this node is lost.
   We set the node as no longer connected. The Cluster Cron will detect
   this connection and will try to get it connected again.

   Instead if the node is a temporary node used to accept a query, we
   completely free the node on error. */
void handleLinkIOError(clusterLink *link) {
    freeClusterLink(link);
}

/* Send the messages queued for the link. */
void clusterWriteHandler(connection *conn) {
    clusterLink *link = connGetPrivateData(conn);
    ssize_t nwritten;
    size_t totwritten = 0;

    while (totwritten < NET_MAX_WRITES_PER_EVENT && listLength(link->send_msg_queue) > 0) {
        listNode *head = listFirst(link->send_msg_queue);
        clusterMsgSendBlock *msgblock = (clusterMsgSendBlock *)head->value;
        clusterMsg *msg = &msgblock->msg;
        size_t msg_offset = link->head_msg_send_offset;
        size_t msg_len = ntohl(msg->totlen);

        nwritten = connWrite(conn, (char *)msg + msg_offset, msg_len - msg_offset);
        if (nwritten <= 0) {
            serverLog(LL_DEBUG, "I/O error writing to node link: %s",
                      (nwritten == -1) ? connGetLastError(conn) : "short write");
            handleLinkIOError(link);
            return;
        }
        if (msg_offset + nwritten < msg_len) {
            /* If full message wasn't written, record the offset
             * and continue sending from this point next time */
            link->head_msg_send_offset += nwritten;
            return;
        }
        serverAssert((msg_offset + nwritten) == msg_len);
        link->head_msg_send_offset = 0;

        /* Delete the node and update our memory tracking */
        uint32_t blocklen = msgblock->totlen;
        listDelNode(link->send_msg_queue, head);
        server.stat_cluster_links_memory -= sizeof(listNode);
        link->send_msg_queue_mem -= sizeof(listNode) + blocklen;