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

Code before:
    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. */
    dictRandomIterator *random_iter = dictGetRandomIterator(server.cluster->nodes);
    while ((de = dictRandomIterNext(random_iter)) != NULL) {
        clusterNode *node = dictGetVal(de);
        /* We free the inbound or outboud link to the node if the link has an
         * oversized message send queue and immediately try reconnecting. */
        clusterNodeCronFreeLinkOnBufferLimitReached(node);
        /* The protocol is that function(s) below return non-zero if the node was
         * terminated.
         */
        if (clusterNodeCronHandleReconnect(node, now, &cluster_node_conn_attempts)) continue;
    }
    dictReleaseRandomIterator(random_iter);
    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++) {
            de = dictGetRandomKey(server.cluster->nodes);
            clusterNode *this = dictGetVal(de);

            /* Don't ping nodes disconnected or with a ping currently active. */
            if (this->link == NULL || this->ping_sent != 0) continue;
            if (this->flags & (CLUSTER_NODE_MYSELF | CLUSTER_NODE_HANDSHAKE)) continue;
            if (min_pong_node == NULL || min_pong > this->pong_received) {
                min_pong_node = this;
                min_pong = this->pong_received;
            }
        }
        if (min_pong_node) {
            serverLog(LL_DEBUG, "Pinging node %.40s (%s)", min_pong_node->name, min_pong_node->human_nodename);
            clusterSendPing(min_pong_node->link, CLUSTERMSG_TYPE_PING);
        }
    }

    /* Iterate nodes to check if we need to flag something as failing.
     * This loop is also responsible to:

Review comment (bug):
I like this idea but if I think about the worst case, some `unlucky nodes` may never get reconnected for a long time (many cron rounds). Please correct me if it's wrong

e.g., 
```
Nodes:  [ N1 N2 N3 N4 ... N2000 ]
Status: [ O O O O ... O X X ]

O = reconnect attempt issued
X = not yet attempted
```

Code after:
        serverLog(LL_WARNING, "Clusterbus handshake timeout %s:%d", node->ip, node->cport);
        clusterDelNode(node);
        return 1;
    }
    if (nodeInNormalState(node) && node->link != NULL && node->inbound_link == NULL &&
        now - node->inbound_link_freed_time > getHandshakeTimeout() &&
        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;