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

Code before:
     * for which we have no address. */
    if (node->flags & (CLUSTER_NODE_MYSELF | CLUSTER_NODE_NOADDR)) return 1;

    if (node->flags & CLUSTER_NODE_PFAIL) server.cluster->stats_pfail_nodes++;

    /* A Node in HANDSHAKE state has a limited lifespan equal to the
     * configured node timeout. */
    if (nodeInHandshake(node) && now - node->ctime > handshake_timeout) {
        clusterDelNode(node);
        return 1;
    }

    if (node->link == NULL) {
        clusterLink *link = createClusterLink(node);
        link->conn = connCreate(connTypeOfCluster());
        connSetPrivateData(link->conn, link);
        if (connConnect(link->conn, node->ip, node->cport, server.bind_source_addr, 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;
    }

    unsigned long long mem_link = link->send_msg_queue_mem;
    if (mem_link > server.cluster_link_msg_queue_limit_bytes) {
        serverLog(LL_WARNING,
                  "Freeing cluster link(%s node %.40s, used memory: %llu) due to "
                  "exceeding send buffer memory limit.",
                  link->inbound ? "from" : "to", link->node ? link->node->name : "", mem_link);
        freeClusterLink(link);
        server.cluster->stat_cluster_links_buffer_limit_exceeded++;
    }
}

Review comment (bug):
Should we rather invoke `resetManualFailover` and clean up `failover_auth_time` in that period?

We should anyway cleanup failover_auth_time in the resetManualFailover method.

Code after:
}

/* If a manual failover timed out, abort it. */
void manualFailoverCheckTimeout(void) {
    if (server.cluster->mf_end && server.cluster->mf_end < mstime()) {
        serverLog(LL_WARNING, "Manual failover timed out.");
        resetManualFailover();
    }
}

/* This function is called from the cluster cron function in order to go
 * forward with a manual failover state machine. */
void clusterHandleManualFailover(void) {
    /* Return ASAP if no manual failover is in progress. */
    if (server.cluster->mf_end == 0) return;

    /* If mf_can_start is non-zero, the failover was already triggered so the
     * next steps are performed by clusterHandleReplicaFailover(). */
    if (server.cluster->mf_can_start) return;

    if (server.cluster->mf_primary_offset == -1) return; /* Wait for offset... */

    if (server.cluster->mf_primary_offset == replicationGetReplicaOffset()) {
        /* Our replication offset matches the primary replication offset
         * announced after clients were paused. We can start the failover. */
        manualFailoverCanStart();
        serverLog(LL_NOTICE, "All primary replication stream processed, "
                             "manual failover can start.");
        clusterDoBeforeSleep(CLUSTER_TODO_HANDLE_FAILOVER);
        return;
    }
    clusterDoBeforeSleep(CLUSTER_TODO_HANDLE_MANUALFAILOVER);
}

/* -----------------------------------------------------------------------------
 * CLUSTER cron job
 * -------------------------------------------------------------------------- */

/* Check if the node is disconnected and re-establish the connection.
 * Also update a few stats while we are here, that can be used to make
 * better decisions in other part of the code. */
static int clusterNodeCronHandleReconnect(clusterNode *node, mstime_t handshake_timeout, mstime_t now) {
    /* Not interested in reconnecting the link with myself or nodes
     * for which we have no address. */
    if (node->flags & (CLUSTER_NODE_MYSELF | CLUSTER_NODE_NOADDR)) return 1;

    if (node->flags & CLUSTER_NODE_PFAIL) server.cluster->stats_pfail_nodes++;

    /* A Node in HANDSHAKE state has a limited lifespan equal to the
     * configured node timeout. */
    if (nodeInHandshake(node) && now - node->ctime > handshake_timeout) {