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

Code before:
        serverLog(LL_NOTICE, "Node hard reset, now I'm %.40s", myself->name);
    }

    /* Re-populate shards */
    clusterAddNodeToShard(myself->shard_id, myself);

    /* Make sure to persist the new config and update the state. */
    clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG | CLUSTER_TODO_UPDATE_STATE | CLUSTER_TODO_FSYNC_CONFIG);
}

/* -----------------------------------------------------------------------------
 * CLUSTER communication link
 * -------------------------------------------------------------------------- */
static void *createClusterMsgSendBlock(int type, uint32_t msglen) {
    uint32_t blocklen;
    if (type == CLUSTERMSG_TYPE_PUBLISH_LIGHT || type == CLUSTERMSG_TYPE_PUBLISHSHARD_LIGHT) {
        blocklen = msglen + sizeof(clusterMsgSendBlockLight) - sizeof(clusterMsgLight);
    } else {
        blocklen = msglen + sizeof(clusterMsgSendBlock) - sizeof(clusterMsg);
    }
    clusterMsgSendBlock *msgblock = zcalloc(blocklen);
    msgblock->refcount = 1;
    msgblock->totlen = blocklen;
    server.stat_cluster_links_memory += blocklen;
    clusterBuildMessageHdr(&msgblock->msg, type, msglen);
    return (void *)msgblock;
}

static void clusterMsgSendBlockDecrRefCount(void *node) {
    clusterMsgSendBlock *msgblock = (clusterMsgSendBlock *)node;
    msgblock->refcount--;
    serverAssert(msgblock->refcount >= 0);
    if (msgblock->refcount == 0) {
        server.stat_cluster_links_memory -= msgblock->totlen;
        zfree(msgblock);
    }
}

clusterLink *createClusterLink(clusterNode *node) {
    clusterLink *link = zmalloc(sizeof(*link));
    link->ctime = mstime();
    link->send_msg_queue = listCreate();
    listSetFreeMethod(link->send_msg_queue, clusterMsgSendBlockDecrRefCount);
    link->head_msg_send_offset = 0;
    link->send_msg_queue_mem = sizeof(list);
    link->rcvbuf = zmalloc(link->rcvbuf_alloc = RCVBUF_INIT_LEN);
    link->rcvbuf_len = 0;
    server.stat_cluster_links_memory += link->rcvbuf_alloc + link->send_msg_queue_mem;
    link->conn = NULL;
    link->node = node;
    /* Related node can only possibly be known at link creation time if this is an outbound link */

Review comment (suggestion):
Passing around `void *` is not good for type safety. Consider unions or other methods.

Code after:
                  "Cluster communication port is 10,000 port "
                  "numbers higher than your %s port. "
                  "Your %s port number must be 55535 or less.",
                  SERVER_TITLE, SERVER_TITLE, SERVER_TITLE);
        exit(1);
    }
    if (!server.bindaddr_count) {
        serverLog(LL_WARNING, "No bind address is configured, but it is required for the Cluster bus.");
        exit(1);
    }

    /* Register our own rdb aux fields */
    serverAssert(rdbRegisterAuxField("cluster-slot-states", clusterEncodeOpenSlotsAuxField,
                                     clusterDecodeOpenSlotsAuxField) == C_OK);

    /* Set myself->port/cport/pport to my listening ports, we'll just need to
     * discover the IP address via MEET messages. */
    deriveAnnouncedPorts(&myself->tcp_port, &myself->tls_port, &myself->cport);

    server.cluster->mf_end = 0;
    server.cluster->mf_replica = NULL;
    for (int conn_type = 0; conn_type < CACHE_CONN_TYPE_MAX; conn_type++) {
        server.cached_cluster_slot_info[conn_type] = NULL;
    }
    resetManualFailover();
    clusterUpdateMyselfFlags();
    clusterUpdateMyselfIp();
    clusterUpdateMyselfClientIpV4();
    clusterUpdateMyselfClientIpV6();
    clusterUpdateMyselfHostname();
    clusterUpdateMyselfHumanNodename();
}

void clusterInitLast(void) {
    if (connectionIndexByType(connTypeOfCluster()->get_type(NULL)) < 0) {
        serverLog(LL_WARNING, "Missing connection type %s, but it is required for the Cluster bus.",
                  connTypeOfCluster()->get_type(NULL));
        exit(1);
    }

    int port = defaultClientPort();
    connListener *listener = &server.clistener;
    listener->count = 0;
    listener->bindaddr = server.bindaddr;
    listener->bindaddr_count = server.bindaddr_count;
    listener->port = server.cluster_port ? server.cluster_port : port + CLUSTER_PORT_INCR;
    listener->ct = connTypeOfCluster();
    if (connListen(listener) == C_ERR) {
        /* Note: the following log text is matched by the test suite. */
        serverLog(LL_WARNING, "Failed listening on port %u (cluster), aborting.", listener->port);
        exit(1);