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

Code before:
    if (createSocketAcceptHandler(&server.clistener, clusterAcceptHandler) != C_OK) {
        serverPanic("Unrecoverable error creating Cluster socket accept handler.");
    }
}

/* Called when a cluster node receives SHUTDOWN. */
void clusterHandleServerShutdown(void) {
    if (server.auto_failover_on_shutdown) {
        /* Find the first best replica, that is, the replica with the largest offset. */
        client *best_replica = NULL;
        listIter replicas_iter;
        listNode *replicas_list_node;
        listRewind(server.replicas, &replicas_iter);
        while ((replicas_list_node = listNext(&replicas_iter)) != NULL) {
            client *replica = listNodeValue(replicas_list_node);
            /* This is done only when the replica offset is caught up, to avoid data loss */
            if (replica->repl_state == REPLICA_STATE_ONLINE && replica->repl_ack_off == server.primary_repl_offset) {
                best_replica = replica;
                break;
            }
        }

        if (best_replica) {
            /* Send a CLUSTER FAILOVER FORCE to the best replica. */
            const char *buf = "*3\r\n$7\r\nCLUSTER\r\n$8\r\nFAILOVER\r\n$5\r\nFORCE\r\n";
            if (connWrite(best_replica->conn, buf, strlen(buf)) == (int)strlen(buf)) {
                serverLog(LL_NOTICE, "Sending CLUSTER FAILOVER FORCE to replica %s succeeded.",
                          replicationGetReplicaName(best_replica));
            } else {
                serverLog(LL_WARNING, "Failed to send CLUSTER FAILOVER FORCE to replica: %s", strerror(errno));
            }
        } else {
            serverLog(LL_NOTICE, "Unable to find a replica to perform an auto failover on shutdown.");
        }
    }

    /* The error logs have been logged in the save function if the save fails. */
    serverLog(LL_NOTICE, "Saving the cluster configuration file before exiting.");
    clusterSaveConfig(1);

#if !defined(__sun)
    /* Unlock the cluster config file before shutdown, see clusterLockConfig.
     *
     * This is needed if you shutdown a very large server process, it will take
     * a while for the OS to release resources and unlock the cluster configuration
     * file. Therefore, if we immediately try to restart the server process, it
     * may not be able to acquire the lock on the cluster configuration file and
     * fail to start. We explicitly releases the lock on the cluster configuration
     * file on shutdown, rather than relying on the OS to release the lock, which
     * is a cleaner and safer way to release acquired resources. */
    if (server.cluster_config_file_lock_fd != -1) {

Review comment (bug):
ok, i think here we have a problem, the primary is sending a CLUSTER FAILOVER to replica, replica will feed it to the replication stream.
```
    /* If the client is a primary we need to compute the difference
     * between the applied offset before and after processing the buffer,
     * to understand how much of the replication stream was actually
     * applied to the primary state: this quantity, and its corresponding
     * part of the replication stream, will be propagated to the
     * sub-replicas and to the replication backlog. */
    if (c->flag.primary) {
        long long applied = c->reploff - prev_offset;
        if (applied) {
            replicationFeedStreamFromPrimaryStream(c->querybuf + c->repl_applied, applied);
            c->repl_applied += applied;
        }
    }
```

I always though a ADMIN command won't do this thing, look like i get it wrong all the time. Do you guys think there is an another way to do this?

Code after:
    }
    if (saveconf) clusterSaveConfigOrDie(1);

    /* Port sanity check II
     * The other handshake port check is triggered too late to stop
     * us from trying to use a too-high cluster port number. */
    int port = defaultClientPort();
    if (!server.cluster_port && port > (65535 - CLUSTER_PORT_INCR)) {
        serverLog(LL_WARNING,
                  "%s port number too high. "
                  "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();
    resetClusterStats();
}

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);
    }