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

Code before:
     */
    return min_nodes_for_coverage * NODE_CONNECTION_RETRIES_PER_TIMEOUT;
}

/* This is executed 10 times every second */
void clusterCron(void) {
    dictIterator *di;
    dictEntry *de;
    int update_state = 0;
    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. */
    di = dictGetSafeIterator(server.cluster->nodes);
    const long long max_conn_attempts = maxConnectionAttemptsPerCron(dictSize(server.cluster->nodes), server.cluster_node_timeout);
    while ((de = dictNext(di)) != 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 (!server.debug_cluster_disable_reconnection && clusterNodeCronHandleReconnect(node, now, &cluster_node_conn_attempts, max_conn_attempts)) continue;
    }
    cluster_node_conn_attempts = 0;
    dictReleaseIterator(di);

    /* 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. */

Review comment (suggestion):
Should this be server.cluster_node_timeout/2, we try to disconnect and reconnect at half the period of node timeout.

```suggestion
    const long long max_conn_attempts = maxConnectionAttemptsPerCron(dictSize(server.cluster->nodes), server.cluster_node_timeout/2);
```

Code after:
}

/* Free outbound link to a node if its send buffer size exceeded limit. */
static void clusterNodeCronFreeLinkOnBufferLimitReached(clusterNode *node) {
    freeClusterLinkOnBufferLimitReached(node->link);
    freeClusterLinkOnBufferLimitReached(node->inbound_link);
}

/* Compute the maximum number of connection attempts the clusterCron
 * loop should schedule in a single cron.
 *
 * We want to guarantee that every node is contacted 10 times within node timeout. */
static long long maxConnectionAttemptsPerCron(void) {
    long long reconnect_interval = server.cluster_node_timeout / 2;
    if (reconnect_interval <= 0)
        return 0;
    /* We run the cron loop every 100 ms.  To reach 100 % of the nodes
     * within the timeout, we need: ceil(nodes * 100 / reconnect_interval) */
    const long long min_nodes_for_coverage = dictSize(server.cluster->nodes) * CLUSTER_CRON_PERIOD_MS / reconnect_interval;
    /* Increase the coverage budget so each node can be probed 10 times
     * inside the timeout. */
    return min_nodes_for_coverage * NODE_CONNECTION_RETRIES_PER_TIMEOUT;
}

/* This is executed 10 times every second */
void clusterCron(void) {
    dictIterator *di;
    dictEntry *de;
    int update_state = 0;
    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;
    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. */
    di = dictGetSafeIterator(server.cluster->nodes);
    long long cluster_node_conn_attempts = maxConnectionAttemptsPerCron();
    while ((de = dictNext(di)) != 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.