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

Code before:
                force = 1; /* Takeover also implies force. */
            } else if (!strcasecmp(c->argv[j]->ptr, "replicaid") && moreargs) {
                j++;
                replicaid = c->argv[j];
            } else {
                addReplyErrorObject(c, shared.syntaxerr);
                return 1;
            }
        }

        /* Check if it should be executed by myself. */
        if (replicaid != NULL) {
            clusterNode *n = clusterLookupNode(replicaid->ptr, sdslen(replicaid->ptr));
            if (n != myself) {
                /* Ignore this command, including the sanity check and the process. */
                addReply(c, shared.ok);
                return 1;
            }
        }

        /* Check preconditions. */
        if (clusterNodeIsPrimary(myself)) {
            if (replicaid == NULL) addReplyError(c, "You should send CLUSTER FAILOVER to a replica");
            return 1;
        } else if (myself->replicaof == NULL) {
            if (replicaid == NULL) addReplyError(c, "I'm a replica but my master is unknown to me");
            return 1;
        } else if (!force && (nodeFailed(myself->replicaof) || myself->replicaof->link == NULL)) {
            if (replicaid == NULL) addReplyError(c, "Master is down or failed, please use CLUSTER FAILOVER FORCE");
            return 1;
        }
        resetManualFailover();
        server.cluster->mf_end = mstime() + CLUSTER_MF_TIMEOUT;
        sds client = catClientInfoShortString(sdsempty(), c, server.hide_user_data_from_log);

        if (takeover) {
            /* A takeover does not perform any initial check. It just
             * generates a new configuration epoch for this node without
             * consensus, claims the primary's slots, and broadcast the new
             * configuration. */
            serverLog(LL_NOTICE, "Taking over the primary (user request from '%s').", client);
            clusterBumpConfigEpochWithoutConsensus();
            clusterFailoverReplaceYourPrimary();
        } else if (force) {
            /* If this is a forced failover, we don't need to talk with our
             * primary to agree about the offset. We just failover taking over
             * it without coordination. */
            if (c == server.primary) {
                serverLog(LL_NOTICE, "Forced failover primary request accepted (primary request from '%s').", client);
            } else {
                serverLog(LL_NOTICE, "Forced failover user request accepted (user request from '%s').", client);

Review comment (bug):
Why add this? If the primary is unknown, the failover can't succeed, so I think we need to return an error even if REPLICAID is sent.

Code after:
                zfree(slots);
                return 1;
            }
            if ((endslot = getSlotOrReply(c, c->argv[j + 1])) == -1) {
                zfree(slots);
                return 1;
            }
            if (startslot > endslot) {
                addReplyErrorFormat(c, "start slot number %d is greater than end slot number %d", startslot, endslot);
                zfree(slots);
                return 1;
            }

            if (checkSlotAssignmentsOrReply(c, slots, del, startslot, endslot) == C_ERR) {
                zfree(slots);
                return 1;
            }
        }
        clusterUpdateSlots(c, slots, del);
        zfree(slots);
        clusterDoBeforeSleep(CLUSTER_TODO_UPDATE_STATE | CLUSTER_TODO_SAVE_CONFIG);
        addReply(c, shared.ok);
    } else if (!strcasecmp(c->argv[1]->ptr, "setslot") && c->argc >= 4) {
        /* SETSLOT 10 MIGRATING <node ID> */
        /* SETSLOT 10 IMPORTING <node ID> */
        /* SETSLOT 10 STABLE */
        /* SETSLOT 10 NODE <node ID> */
        clusterCommandSetSlot(c);
    } else if (!strcasecmp(c->argv[1]->ptr, "bumpepoch") && c->argc == 2) {
        /* CLUSTER BUMPEPOCH */
        int retval = clusterBumpConfigEpochWithoutConsensus();
        sds reply = sdscatprintf(sdsempty(), "+%s %llu\r\n", (retval == C_OK) ? "BUMPED" : "STILL",
                                 (unsigned long long)myself->configEpoch);
        addReplySds(c, reply);
    } else if (!strcasecmp(c->argv[1]->ptr, "saveconfig") && c->argc == 2) {
        int retval = clusterSaveConfig(1);

        if (retval == C_OK)
            addReply(c, shared.ok);
        else
            addReplyErrorFormat(c, "error saving the cluster node config: %s", strerror(errno));
    } else if (!strcasecmp(c->argv[1]->ptr, "forget") && c->argc == 3) {
        /* CLUSTER FORGET <NODE ID> */
        clusterNode *n = clusterLookupNode(c->argv[2]->ptr, sdslen(c->argv[2]->ptr));
        if (!n) {
            if (clusterBlacklistExists((char *)c->argv[2]->ptr))
                /* Already forgotten. The deletion may have been gossipped by
                 * another node, so we pretend it succeeded. */
                addReply(c, shared.ok);
            else
                addReplyErrorFormat(c, "Unknown node %s", (char *)c->argv[2]->ptr);