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

Code before:
    int orphaned_primaries; /* How many primaries there are without ok replicas. */
    int max_replicas;       /* Max number of ok replicas for a single primary. */
    int this_replicas;      /* Number of ok replicas for our primary (if we are replica). */
    mstime_t min_pong = 0, now = mstime();
    clusterNode *min_pong_node = NULL;
    static unsigned long long iteration = 0;
    int cluster_node_conn_attempts = 0;

    iteration++; /* Number of times this function was called so far. */

    clusterUpdateMyselfHostname();

    /* Clear so clusterNodeCronHandleReconnect can count the number of nodes in PFAIL. */
    server.cluster->stats_pfail_nodes = 0;
    /* Run through some of the operations we want to do on each cluster node. */
    /* We want to start from the random position within cluster nodes just to be safe. */
    int total = dictSize(server.cluster->nodes);
    int skip = rand() % (total == 0 ? 1 : total);

    /* First iterator: from skip point to end */
    di = dictGetSafeIterator(server.cluster->nodes);
    int i = 0;

    /* Skip first 'skip' entries */
    while (i++ < skip && (de = dictNext(di)) != NULL) {
    }

    while ((de = dictNext(di)) != NULL) {
        clusterNode *node = dictGetVal(de);
        handleClusterNode(node, now, &cluster_node_conn_attempts);
    }
    dictReleaseIterator(di);

    /* Second iterator: from start to skip point */
    di = dictGetSafeIterator(server.cluster->nodes);
    i = 0;
    while (i++ < skip && (de = dictNext(di)) != NULL) {
        clusterNode *node = dictGetVal(de);
        handleClusterNode(node, now, &cluster_node_conn_attempts);
    }
    dictReleaseIterator(di);
    cluster_node_conn_attempts = 0;

    /* Ping some random node 1 time every 10 iterations, so that we usually ping
     * one random node every second. */
    if (!server.debug_cluster_disable_random_ping && !(iteration % 10)) {
        int j;

        /* Check a few random nodes and ping the one with the oldest
         * pong_received time. */
        for (j = 0; j < 5; j++) {

Review comment (suggestion):
I would prefer us taking this functionality to the dictionary level to avoid these random pointer access. It would be nice if we can introduce an API like `dictGetRandomIterator(d)` and avoid maintaining any information on the caller side.

Code after:
        now - node->meet_sent > getHandshakeTimeout()) {
        /* Node has an outbound link, but no inbound link for more than the handshake timeout.
         * This probably means this node does not know us yet, whereas we know it.
         * So we send it a MEET packet to do a handshake with it and correct the inconsistent cluster view.
         * We make sure to not re-send a MEET packet more than once every handshake timeout period, so as to
         * leave the other node time to complete the handshake. */
        node->flags |= CLUSTER_NODE_MEET;
        serverLog(LL_NOTICE, "Sending MEET packet to node %.40s (%s) because there is no inbound link for it",
                  node->name, node->human_nodename);
        clusterSendPing(node->link, CLUSTERMSG_TYPE_MEET);
    }

    if (node->link == NULL) {
        mstime_t reconnect_interval = server.cluster_node_timeout / 2;
        /* Skip this outbound connection attempt when all these conditions are true:
         *  1. No inbound link from the peer exists.
         *  2. The back‑off window since the last try is still active
         *  3. The node has already exceeded its retry budget for this cron cycle
         */
        if (!node->inbound_link && (now - node->outbound_link_attempt_time < reconnect_interval / NODE_CONNECTION_RETRIES_PER_TIMEOUT && *cluster_conn_attempts == 0)) {
            return 1;
        }
        node->outbound_link_attempt_time = now;
        (*cluster_conn_attempts)--;
        clusterLink *link = createClusterLink(node);
        link->conn = connCreate(connTypeOfCluster());
        connSetPrivateData(link->conn, link);
        if (connConnect(link->conn, node->ip, node->cport, server.bind_source_addr, 0, clusterLinkConnectHandler) ==
            C_ERR) {
            /* We got a synchronous error from connect before
             * clusterSendPing() had a chance to be called.
             * If node->ping_sent is zero, failure detection can't work,
             * so we claim we actually sent a ping now (that will
             * be really sent as soon as the link is obtained). */
            if (node->ping_sent == 0) node->ping_sent = mstime();
            serverLog(LL_DEBUG,
                      "Unable to connect to "
                      "Cluster Node [%s]:%d -> %s",
                      node->ip, node->cport, server.neterr);

            freeClusterLink(link);
            return 0;
        }
    }
    return 0;
}

static void freeClusterLinkOnBufferLimitReached(clusterLink *link) {
    if (link == NULL || server.cluster_link_msg_queue_limit_bytes == 0) {
        return;
    }