src/cluster.c - unstable - githubmirror/redis

src/cluster.c
2 501 строка94 KB
Yuan Wang
Move common topology change handling into the cluster layer (#15197)
07 авг 2026, 04:38
07 авг 2026, 04:385d67a63
/*
 * Copyright (c) 2009-Present, Redis Ltd.
 * All rights reserved.
 *
 * Copyright (c) 2024-present, Valkey contributors.
 * All rights reserved.
 *
 * Licensed under your choice of (a) the Redis Source Available License 2.0
 * (RSALv2); or (b) the Server Side Public License v1 (SSPLv1); or (c) the
 * GNU Affero General Public License v3 (AGPLv3).
 *
 * Portions of this file are available under BSD3 terms; see REDISCONTRIBUTIONS for more information.
 */

/*
 * cluster.c contains the common parts of a clustering
 * implementation, the parts that are shared between
 * any implementation of Redis clustering.
 */

#include "server.h"
#include "cluster.h"
#include "cluster_asm.h"
#include "cluster_slot_stats.h"

#include <ctype.h>
#include "bio.h"

/* -----------------------------------------------------------------------------
 * Key space handling
 * -------------------------------------------------------------------------- */

/* If it can be inferred that the given glob-style pattern, as implemented in
 * stringmatchlen() in util.c, only can match keys belonging to a single slot,
 * that slot is returned. Otherwise -1 is returned. */
int patternHashSlot(char *pattern, int length) {
    int s = -1; /* index of the first '{' */

    for (int i = 0; i < length; i++) {
        if (pattern[i] == '*' || pattern[i] == '?' || pattern[i] == '[') {
            /* Wildcard or character class found. Keys can be in any slot. */
            return -1;
        } else if (pattern[i] == '\\') {
            /* Escaped character. Computing slot in this case is not
             * implemented. We would need a temp buffer. */
            return -1;
        } else if (s == -1 && pattern[i] == '{') {
            /* Opening brace '{' found. */
            s = i;
        } else if (s >= 0 && pattern[i] == '}' && i == s + 1) {
            /* Empty tag '{}' found. The whole key is hashed. Ignore braces. */
            s = -2;
        } else if (s >= 0 && pattern[i] == '}') {
            /* Non-empty tag '{...}' found. Hash what's between braces. */
            return crc16(pattern + s + 1, i - s - 1) & 0x3FFF;
        }
    }

    /* The pattern matches a single key. Hash the whole pattern. */
    return crc16(pattern, length) & 0x3FFF;
}

int getSlotOrReply(client *c, robj *o) {
    long long slot;

    if (getLongLongFromObject(o,&slot) != C_OK ||
        slot < 0 || slot >= CLUSTER_SLOTS)
    {
        addReplyError(c,"Invalid or out of range slot");
        return -1;
    }
    return (int) slot;
}

ConnectionType *connTypeOfCluster(void) {
    if (server.tls_cluster) {
        return connectionTypeTls();
    }

    return connectionTypeTcp();
}

/* -----------------------------------------------------------------------------
 * DUMP, RESTORE and MIGRATE commands
 * -------------------------------------------------------------------------- */

/* Generates a DUMP-format representation of the object 'o', adding it to the
 * io stream pointed by 'rio'. Flags can omit the checksum or key metadata.
 * This function can't fail. */
void createDumpPayload(rio *payload, robj *o, robj *key, int dbid, int flags, size_t size_hint) {
    unsigned char buf[2];
    uint64_t crc = 0;

    /* Serialize the object in an RDB-like format. It consist of an object type
     * byte followed by the serialized object. This is understood by RESTORE.
     * 'size_hint', when non-zero, sizes the buffer to the caller's estimate of
     * the payload in a single allocation to avoid reallocations. */
    sds buffer;
    if (size_hint) {
        buffer = sdsnewlen(SDS_NOINIT, size_hint);
        sdssetlen(buffer, 0);
        buffer[0] = '\0';
    } else {
        buffer = sdsempty();
    }
    rioInitWithBuffer(payload,buffer);
    payload->flags |= RIO_FLAG_DUMP_PAYLOAD;

    /* Skip compression when the caller asked for raw bytes. */
    int prev_comp = server.rdb_compression;
    if (flags & DUMP_PAYLOAD_DONT_COMPRESS) server.rdb_compression = 0;

    /* Save key metadata if present (TTL is handled separately via command
     * args). AOF RESTORE payloads omit it because AOF rewrite handles module
     * metadata separately through keyMetaOnAof(). */
    if (!(flags & DUMP_PAYLOAD_SKIP_KEY_META) && getModuleMetaBits(o->metabits))
        serverAssert(rdbSaveKeyMetadata(payload, key, o, dbid) != -1);
    serverAssert(rdbSaveObjectType(payload,o));
    serverAssert(rdbSaveObject(payload,o,key,dbid));

    server.rdb_compression = prev_comp;

    /* Write the footer, this is how it looks like:
     * ----------------+---------------------+---------------+
     * ... RDB payload | 2 bytes RDB version | 8 bytes CRC64 |
     * ----------------+---------------------+---------------+
     * RDB version and CRC are both in little endian.
     */

    /* RDB version */
    buf[0] = RDB_VERSION & 0xff;
    buf[1] = (RDB_VERSION >> 8) & 0xff;
    payload->io.buffer.ptr = sdscatlen(payload->io.buffer.ptr,buf,2);

    /* If crc checksum is disabled, crc is set to 0 and no checksum validation
     * will be performed on RESTORE. */
    if (!(flags & DUMP_PAYLOAD_SKIP_CHECKSUM)) {
        /* CRC64 */
        crc = crc64(0,(unsigned char*)payload->io.buffer.ptr,
                    sdslen(payload->io.buffer.ptr));
        memrev64ifbe(&crc);
    }
    payload->io.buffer.ptr = sdscatlen(payload->io.buffer.ptr,&crc,8);
}

/* CRC-less, compression-less DUMP payload for RESTORE; returns an sds the caller
 * owns. size_hint (if nonzero) presizes. */
sds createRawDumpPayload(robj *o, robj *key, int dbid, int flags, size_t size_hint) {
    rio payload;
    createDumpPayload(&payload, o, key, dbid,
                      DUMP_PAYLOAD_SKIP_CHECKSUM | DUMP_PAYLOAD_DONT_COMPRESS | flags, size_hint);
    return payload.io.buffer.ptr;
}

/* Verify that the RDB version of the dump payload matches the one of this Redis
 * instance and that the checksum is ok.
 * If the DUMP payload looks valid C_OK is returned, otherwise C_ERR
 * is returned. If rdbver_ptr is not NULL, its populated with the value read
 * from the input buffer. */
int verifyDumpPayload(unsigned char *p, size_t len, uint16_t *rdbver_ptr) {
    unsigned char *footer;
    uint16_t rdbver;
    uint64_t crc;

    /* At least 2 bytes of RDB version and 8 of CRC64 should be present. */
    if (len < 10) return C_ERR;
    footer = p+(len-10);

    /* Set and verify RDB version. */
    rdbver = (footer[1] << 8) | footer[0];
    if (rdbver_ptr) {
        *rdbver_ptr = rdbver;
    }
    if (rdbver > RDB_VERSION) return C_ERR;

    if (server.skip_checksum_validation)
        return C_OK;

    uint64_t crc_payload;
    memcpy(&crc_payload, footer+2, 8);
    if (crc_payload == 0) /* No checksum. */
        return C_OK;

    /* Verify CRC64 */
    crc = crc64(0,p,len-8);
    memrev64ifbe(&crc);
    return crc == crc_payload ? C_OK : C_ERR;
}

/* DUMP keyname
 * DUMP is actually not used by Redis Cluster but it is the obvious
 * complement of RESTORE and can be useful for different applications. */
void dumpCommand(client *c) {
    kvobj *o;
    rio payload;

    /* Check if the key is here. */
    if ((o = lookupKeyRead(c->db,c->argv[1])) == NULL) {
        addReplyNull(c);
        return;
    }

    /* Create the DUMP encoded representation. */
    createDumpPayload(&payload,o,c->argv[1],c->db->id,0,0);

    /* Transfer to the client */
    addReplyBulkSds(c,payload.io.buffer.ptr);
    return;
}

/* RESTORE key ttl serialized-value [REPLACE] [ABSTTL] [IDLETIME seconds] [FREQ frequency] */
void restoreCommand(client *c) {
    long long ttl, lfu_freq = -1, lru_idle = -1, lru_clock = -1;
    rio payload;
    int j, type, replace = 0, absttl = 0;
    robj *obj;

    /* Parse additional options */
    for (j = 4; j < c->argc; j++) {
        int additional = c->argc-j-1;
        if (!strcasecmp(c->argv[j]->ptr,"replace")) {
            replace = 1;
        } else if (!strcasecmp(c->argv[j]->ptr,"absttl")) {
            absttl = 1;
        } else if (!strcasecmp(c->argv[j]->ptr,"idletime") && additional >= 1 &&
                   lfu_freq == -1)
        {
            if (getLongLongFromObjectOrReply(c,c->argv[j+1],&lru_idle,NULL)
                != C_OK) return;
            if (lru_idle < 0) {
                addReplyError(c,"Invalid IDLETIME value, must be >= 0");
                return;
            }
            lru_clock = LRU_CLOCK();
            j++; /* Consume additional arg. */
        } else if (!strcasecmp(c->argv[j]->ptr,"freq") && additional >= 1 &&
                   lru_idle == -1)
        {
            if (getLongLongFromObjectOrReply(c,c->argv[j+1],&lfu_freq,NULL)
                != C_OK) return;
            if (lfu_freq < 0 || lfu_freq > 255) {
                addReplyError(c,"Invalid FREQ value, must be >= 0 and <= 255");
                return;
            }
            j++; /* Consume additional arg. */
        } else {
            addReplyErrorObject(c,shared.syntaxerr);
            return;
        }
    }

    /* Make sure this key does not already exist here... */
    robj *key = c->argv[1];
    if (!replace && lookupKeyWrite(c->db,key) != NULL) {
        addReplyErrorObject(c,shared.busykeyerr);
        return;
    }

    /* Check if the TTL value makes sense */
    if (getLongLongFromObjectOrReply(c,c->argv[2],&ttl,NULL) != C_OK) {
        return;
    } else if (ttl < 0) {
        addReplyError(c,"Invalid TTL value, must be >= 0");
        return;
    }

    /* Verify RDB version and data checksum. */
    if (verifyDumpPayload(c->argv[3]->ptr,sdslen(c->argv[3]->ptr),NULL) == C_ERR)
    {
        addReplyError(c,"DUMP payload version or checksum are wrong");
        return;
    }

    rioInitWithBuffer(&payload,c->argv[3]->ptr);
    payload.flags |= RIO_FLAG_DUMP_PAYLOAD;

    /* Initialize metadata spec to collect metadata+expiry from payload. */
    KeyMetaSpec keymeta;
    keyMetaSpecInit(&keymeta);

    /* Compute TTL early so we can add it to metadata spec in correct order */
    if (ttl) {
        if (!absttl) ttl+=commandTimeSnapshot();
        keyMetaSpecAdd(&keymeta, KEY_META_ID_EXPIRE, ttl);
    }

    /* With metadata, type = RDB_OPCODE_KEY_META. Layout: [<META>,]<TYPE>,<KEY>,<VALUE> */
    type = rdbLoadType(&payload);
    if (rdbResolveKeyType(&payload, &type, c->db->id, &keymeta) == -1) {
        addReplyError(c,"Bad data format");
        return;
    }

    /* Load the object */
    if ((obj = rdbLoadObject(type,&payload,key->ptr,c->db->id,NULL)) == NULL)
    {
        keyMetaSpecCleanup(&keymeta);
        addReplyError(c,"Bad data format");
        return;
    }

    /* Resolve the key's existence and its insertion link. On the common new-key
     * path dbAddInternal() below reuses the link instead of probing again. */
    dictEntryLink link = NULL;
    kvobj *oldval = lookupKeyWriteWithLink(c->db, key, &link);
    int oldtype = oldval ? oldval->type : -1;

    /* Call dbDelete() only when a key is actually present:
     *   oldval != NULL -> key exists.
     *   link  == NULL  -> an expired key might still be physically present and 
     *                     must be deleted. */
    int deleted = 0;
    if (replace && (oldval || !link)) {
        deleted = dbDelete(c->db,key);
        link = NULL; /* dbDelete invalidated the link */
    }

    if (ttl && checkAlreadyExpired(ttl)) {
        if (deleted) {
            robj *aux = server.lazyfree_lazy_server_del ? shared.unlink : shared.del;
            rewriteClientCommandVector(c, 2, aux, key);
            keyModified(c,c->db,key,NULL,1);
            notifyKeyspaceEvent(NOTIFY_GENERIC,"del",key,c->db->id);
            server.dirty++;
        }
        /* Update the stats, see setGenericCommand for details. */
        server.stat_expiredkeys++;
        keyMetaSpecCleanup(&keymeta);
        decrRefCount(obj);
        addReply(c, shared.ok);
        return;
    }

    /* Create the key and set the TTL if any */
    kvobj *kv = dbAddInternal(c->db, key, &obj, &link, &keymeta);

    /* Save type: kv may be reallocated by module callbacks during notifyKeyspaceEvent below. */
    int kvtype = kv->type;

    /* If minExpiredField was set, then the object is hash with expiration
     * on fields and need to register it in global HFE DS */
    if (kvtype == OBJ_HASH) {
        uint64_t minExpiredField = hashTypeGetMinExpire(kv, 1);
        if (minExpiredField != EB_EXPIRE_TIME_INVALID)
            estoreAdd(c->db->subexpires, getKeySlot(key->ptr), kv, minExpiredField);
    }

    if (kvtype == OBJ_STREAM)
        streamKeyLoaded(c->db, key, kv);

    if (ttl) {
        if (!absttl) {
            /* Propagate TTL as absolute timestamp */
            robj *ttl_obj = createStringObjectFromLongLong(ttl);
            rewriteClientCommandArgument(c,2,ttl_obj);
            decrRefCount(ttl_obj);
            rewriteClientCommandArgument(c,c->argc,shared.absttl);
        }
    }
    objectSetLRUOrLFU(kv, lfu_freq, lru_idle, lru_clock, 1000);
    keyModified(c,c->db,key,NULL,1);
    notifyKeyspaceEvent(NOTIFY_GENERIC,"restore",key,c->db->id);
    KSN_INVALIDATE_KVOBJ(kv);

    /* If we deleted a key that means REPLACE parameter was passed and the
     * destination key existed. */
    if (deleted) {
        notifyKeyspaceEvent(NOTIFY_OVERWRITTEN, "overwritten", key, c->db->id);
        if (oldtype != kvtype) {
            notifyKeyspaceEvent(NOTIFY_TYPE_CHANGED, "type_changed", key, c->db->id);
        }
    }
    addReply(c,shared.ok);
    server.dirty++;
}
/* MIGRATE socket cache implementation.
 *
 * We take a map between host:ip and a TCP socket that we used to connect
 * to this instance in recent time.
 * This sockets are closed when the max number we cache is reached, and also
 * in serverCron() when they are around for more than a few seconds. */
#define MIGRATE_SOCKET_CACHE_ITEMS 64 /* max num of items in the cache. */
#define MIGRATE_SOCKET_CACHE_TTL 10 /* close cached sockets after 10 sec. */

typedef struct migrateCachedSocket {
    connection *conn;
    long last_dbid;
    time_t last_use_time;
} migrateCachedSocket;

/* Return a migrateCachedSocket containing a TCP socket connected with the
 * target instance, possibly returning a cached one.
 *
 * This function is responsible of sending errors to the client if a
 * connection can't be established. In this case -1 is returned.
 * Otherwise on success the socket is returned, and the caller should not
 * attempt to free it after usage.
 *
 * If the caller detects an error while using the socket, migrateCloseSocket()
 * should be called so that the connection will be created from scratch
 * the next time. */
migrateCachedSocket* migrateGetSocket(client *c, robj *host, robj *port, long timeout) {
    connection *conn;
    sds name = sdsempty();
    migrateCachedSocket *cs;

    /* Check if we have an already cached socket for this ip:port pair. */
    name = sdscatlen(name,host->ptr,sdslen(host->ptr));
    name = sdscatlen(name,":",1);
    name = sdscatlen(name,port->ptr,sdslen(port->ptr));
    cs = dictFetchValue(server.migrate_cached_sockets,name);
    if (cs) {
        sdsfree(name);
        cs->last_use_time = server.unixtime;
        return cs;
    }

    /* No cached socket, create one. */
    if (dictSize(server.migrate_cached_sockets) == MIGRATE_SOCKET_CACHE_ITEMS) {
        /* Too many items, drop one at random. */
        dictEntry *de = dictGetRandomKey(server.migrate_cached_sockets);
        cs = dictGetVal(de);
        connClose(cs->conn);
        zfree(cs);
        dictDelete(server.migrate_cached_sockets,dictGetKey(de));
    }

    /* Create the connection */
    conn = connCreate(server.el, connTypeOfCluster());
    if (connBlockingConnect(conn, host->ptr, atoi(port->ptr), timeout)
        != C_OK) {
        addReplyError(c,"-IOERR error or timeout connecting to the client");
        connClose(conn);
        sdsfree(name);
        return NULL;
    }
    connEnableTcpNoDelay(conn);

    /* Add to the cache and return it to the caller. */
    cs = zmalloc(sizeof(*cs));
    cs->conn = conn;

    cs->last_dbid = -1;
    cs->last_use_time = server.unixtime;
    dictAdd(server.migrate_cached_sockets,name,cs);
    return cs;
}

/* Free a migrate cached connection. */
void migrateCloseSocket(robj *host, robj *port) {
    sds name = sdsempty();
    migrateCachedSocket *cs;

    name = sdscatlen(name,host->ptr,sdslen(host->ptr));
    name = sdscatlen(name,":",1);
    name = sdscatlen(name,port->ptr,sdslen(port->ptr));
    cs = dictFetchValue(server.migrate_cached_sockets,name);
    if (!cs) {
        sdsfree(name);
        return;
    }

    connClose(cs->conn);
    zfree(cs);
    dictDelete(server.migrate_cached_sockets,name);
    sdsfree(name);
}

void migrateCloseTimedoutSockets(void) {
    dictIterator di;
    dictEntry *de;

    dictInitSafeIterator(&di, server.migrate_cached_sockets);
    while((de = dictNext(&di)) != NULL) {
        migrateCachedSocket *cs = dictGetVal(de);

        if ((server.unixtime - cs->last_use_time) > MIGRATE_SOCKET_CACHE_TTL) {
            connClose(cs->conn);
            zfree(cs);
            dictDelete(server.migrate_cached_sockets,dictGetKey(de));
        }
    }
    dictResetIterator(&di);
}

/* MIGRATE host port key dbid timeout [COPY | REPLACE | AUTH password |
 *         AUTH2 username password]
 *
 * On in the multiple keys form:
 *
 * MIGRATE host port "" dbid timeout [COPY | REPLACE | AUTH password |
 *         AUTH2 username password] KEYS key1 key2 ... keyN */
void migrateCommand(client *c) {
    migrateCachedSocket *cs;
    int copy = 0, replace = 0, j;
    char *username = NULL;
    char *password = NULL;
    long timeout;
    long dbid;
    robj **kvArray = NULL; /* Objects to migrate. */
    robj **keyArray = NULL; /* Key names. */
    robj **newargv = NULL; /* Used to rewrite the command as DEL ... keys ... */
    rio cmd, payload;
    int may_retry = 1;
    int write_error = 0;
    int argv_rewritten = 0;

    /* To support the KEYS option we need the following additional state. */
    int first_key = 3; /* Argument index of the first key. */
    int num_keys = 1;  /* By default only migrate the 'key' argument. */

    /* Parse additional options */
    for (j = 6; j < c->argc; j++) {
        int moreargs = (c->argc-1) - j;
        if (!strcasecmp(c->argv[j]->ptr,"copy")) {
            copy = 1;
        } else if (!strcasecmp(c->argv[j]->ptr,"replace")) {
            replace = 1;
        } else if (!strcasecmp(c->argv[j]->ptr,"auth")) {
            if (!moreargs) {
                addReplyErrorObject(c,shared.syntaxerr);
                return;
            }
            j++;
            password = c->argv[j]->ptr;
            redactClientCommandArgument(c,j);
        } else if (!strcasecmp(c->argv[j]->ptr,"auth2")) {
            if (moreargs < 2) {
                addReplyErrorObject(c,shared.syntaxerr);
                return;
            }
            username = c->argv[++j]->ptr;
            redactClientCommandArgument(c,j);
            password = c->argv[++j]->ptr;
            redactClientCommandArgument(c,j);
        } else if (!strcasecmp(c->argv[j]->ptr,"keys")) {
            if (sdslen(c->argv[3]->ptr) != 0) {
                addReplyError(c,
                              "When using MIGRATE KEYS option, the key argument"
                              " must be set to the empty string");
                return;
            }
            first_key = j+1;
            num_keys = c->argc - j - 1;
            break; /* All the remaining args are keys. */
        } else {
            addReplyErrorObject(c,shared.syntaxerr);
            return;
        }
    }

    /* Sanity check */
    if (getLongFromObjectOrReply(c,c->argv[5],&timeout,NULL) != C_OK ||
        getLongFromObjectOrReply(c,c->argv[4],&dbid,NULL) != C_OK)
    {
        return;
    }
    if (timeout <= 0) timeout = 1000;

    /* Check if the keys are here. If at least one key is to migrate, do it
     * otherwise if all the keys are missing reply with "NOKEY" to signal
     * the caller there was nothing to migrate. We don't return an error in
     * this case, since often this is due to a normal condition like the key
     * expiring in the meantime. */
    kvArray = zrealloc(kvArray,sizeof(kvobj*)*num_keys);
    keyArray = zrealloc(keyArray,sizeof(robj*)*num_keys);
    int num_exists = 0;

    for (j = 0; j < num_keys; j++) {
        if ((kvArray[num_exists] = lookupKeyRead(c->db,c->argv[first_key+j])) != NULL) {
            keyArray[num_exists] = c->argv[first_key+j];
            num_exists++;
        }
    }
    num_keys = num_exists;
    if (num_keys == 0) {
        zfree(kvArray); zfree(keyArray);
        addReplySds(c,sdsnew("+NOKEY\r\n"));
        return;
    }

    try_again:
    write_error = 0;

    /* Connect */
    cs = migrateGetSocket(c,c->argv[1],c->argv[2],timeout);
    if (cs == NULL) {
        zfree(kvArray); zfree(keyArray);
        return; /* error sent to the client by migrateGetSocket() */
    }

    rioInitWithBuffer(&cmd,sdsempty());

    /* Authentication */
    if (password) {
        int arity = username ? 3 : 2;
        serverAssertWithInfo(c,NULL,rioWriteBulkCount(&cmd,'*',arity));
        serverAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,"AUTH",4));
        if (username) {
            serverAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,username,
                                                           sdslen(username)));
        }
        serverAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,password,
                                                       sdslen(password)));
    }

    /* Send the SELECT command if the current DB is not already selected. */
    int select = cs->last_dbid != dbid; /* Should we emit SELECT? */
    if (select) {
        serverAssertWithInfo(c,NULL,rioWriteBulkCount(&cmd,'*',2));
        serverAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,"SELECT",6));
        serverAssertWithInfo(c,NULL,rioWriteBulkLongLong(&cmd,dbid));
    }

    int non_expired = 0; /* Number of keys that we'll find non expired.
                            Note that serializing large keys may take some time
                            so certain keys that were found non expired by the
                            lookupKey() function, may be expired later. */

    /* Create RESTORE payload and generate the protocol to call the command. */
    for (j = 0; j < num_keys; j++) {
        long long ttl = 0;
        long long expireat = kvobjGetExpire(kvArray[j]);

        if (expireat != -1) {
            ttl = expireat-commandTimeSnapshot();
            if (ttl < 0) {
                continue;
            }
            if (ttl < 1) ttl = 1;
        }

        /* Relocate valid (non expired) keys and values into the array in successive
         * positions to remove holes created by the keys that were present
         * in the first lookup but are now expired after the second lookup. */
        kvArray[non_expired] = kvArray[j];
        keyArray[non_expired++] = keyArray[j];

        serverAssertWithInfo(c,NULL,
                             rioWriteBulkCount(&cmd,'*',replace ? 5 : 4));

        if (server.cluster_enabled)
            serverAssertWithInfo(c,NULL,
                                 rioWriteBulkString(&cmd,"RESTORE-ASKING",14));
        else
            serverAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,"RESTORE",7));
        serverAssertWithInfo(c,NULL,sdsEncodedObject(keyArray[j]));
        serverAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,keyArray[j]->ptr,
                                                       sdslen(keyArray[j]->ptr)));
        serverAssertWithInfo(c,NULL,rioWriteBulkLongLong(&cmd,ttl));

        /* Emit the payload argument, that is the serialized object using
         * the DUMP format. */
        createDumpPayload(&payload,kvArray[j],keyArray[j],dbid,0,0);
        serverAssertWithInfo(c,NULL,
                             rioWriteBulkString(&cmd,payload.io.buffer.ptr,
                                                sdslen(payload.io.buffer.ptr)));
        sdsfree(payload.io.buffer.ptr);

        /* Add the REPLACE option to the RESTORE command if it was specified
         * as a MIGRATE option. */
        if (replace)
            serverAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,"REPLACE",7));
    }

    /* Fix the actual number of keys we are migrating. */
    num_keys = non_expired;

    /* Transfer the query to the other node in 64K chunks. */
    errno = 0;
    {
        sds buf = cmd.io.buffer.ptr;
        size_t pos = 0, towrite;
        int nwritten = 0;

        while ((towrite = sdslen(buf)-pos) > 0) {
            towrite = (towrite > (64*1024) ? (64*1024) : towrite);
            nwritten = connSyncWrite(cs->conn,buf+pos,towrite,timeout);
            if (nwritten != (signed)towrite) {
                write_error = 1;
                goto socket_err;
            }
            pos += nwritten;
        }
    }

    char buf0[1024]; /* Auth reply. */
    char buf1[1024]; /* Select reply. */
    char buf2[1024]; /* Restore reply. */

    /* Read the AUTH reply if needed. */
    if (password && connSyncReadLine(cs->conn, buf0, sizeof(buf0), timeout) <= 0)
        goto socket_err;

    /* Read the SELECT reply if needed. */
    if (select && connSyncReadLine(cs->conn, buf1, sizeof(buf1), timeout) <= 0)
        goto socket_err;

    /* Read the RESTORE replies. */
    int error_from_target = 0;
    int socket_error = 0;
    int del_idx = 1; /* Index of the key argument for the replicated DEL op. */

    /* Allocate the new argument vector that will replace the current command,
     * to propagate the MIGRATE as a DEL command (if no COPY option was given).
     * We allocate num_keys+1 because the additional argument is for "DEL"
     * command name itself. */
    if (!copy) newargv = zmalloc(sizeof(robj*)*(num_keys+1));

    for (j = 0; j < num_keys; j++) {
        if (connSyncReadLine(cs->conn, buf2, sizeof(buf2), timeout) <= 0) {
            socket_error = 1;
            break;
        }
        if ((password && buf0[0] == '-') ||
            (select && buf1[0] == '-') ||
            buf2[0] == '-')
        {
            /* On error assume that last_dbid is no longer valid. */
            if (!error_from_target) {
                cs->last_dbid = -1;
                char *errbuf;
                if (password && buf0[0] == '-') errbuf = buf0;
                else if (select && buf1[0] == '-') errbuf = buf1;
                else errbuf = buf2;

                error_from_target = 1;
                addReplyErrorFormat(c,"Target instance replied with error: %s",
                                    errbuf+1);
            }
        } else {
            if (!copy) {
                /* No COPY option: remove the local key, signal the change. */
                dbDelete(c->db,keyArray[j]);
                keyModified(c,c->db,keyArray[j],NULL,1);
                notifyKeyspaceEvent(NOTIFY_GENERIC,"del",keyArray[j],c->db->id);
                server.dirty++;

                /* Populate the argument vector to replace the old one. */
                newargv[del_idx++] = keyArray[j];
                incrRefCount(keyArray[j]);
            }
        }
    }

    /* On socket error, if we want to retry, do it now before rewriting the
     * command vector. We only retry if we are sure nothing was processed
     * and we failed to read the first reply (j == 0 test). */
    if (!error_from_target && socket_error && j == 0 && may_retry &&
        errno != ETIMEDOUT)
    {
        goto socket_err; /* A retry is guaranteed because of tested conditions.*/
    }

    /* On socket errors, close the migration socket now that we still have
     * the original host/port in the ARGV. Later the original command may be
     * rewritten to DEL and will be too later. */
    if (socket_error) migrateCloseSocket(c->argv[1],c->argv[2]);

    if (!copy) {
        /* Translate MIGRATE as DEL for replication/AOF. Note that we do
         * this only for the keys for which we received an acknowledgement
         * from the receiving Redis server, by using the del_idx index. */
        if (del_idx > 1) {
            newargv[0] = createStringObject("DEL",3);
            /* Note that the following call takes ownership of newargv. */
            replaceClientCommandVector(c,del_idx,newargv);
            argv_rewritten = 1;
        } else {
            /* No key transfer acknowledged, no need to rewrite as DEL. */
            zfree(newargv);
        }
        newargv = NULL; /* Make it safe to call zfree() on it in the future. */
    }

    /* If we are here and a socket error happened, we don't want to retry.
     * Just signal the problem to the client, but only do it if we did not
     * already queue a different error reported by the destination server. */
    if (!error_from_target && socket_error) {
        may_retry = 0;
        goto socket_err;
    }

    if (!error_from_target) {
        /* Success! Update the last_dbid in migrateCachedSocket, so that we can
         * avoid SELECT the next time if the target DB is the same. Reply +OK.
         *
         * Note: If we reached this point, even if socket_error is true
         * still the SELECT command succeeded (otherwise the code jumps to
         * socket_err label. */
        cs->last_dbid = dbid;
        addReply(c,shared.ok);
    } else {
        /* On error we already sent it in the for loop above, and set
         * the currently selected socket to -1 to force SELECT the next time. */
    }

    sdsfree(cmd.io.buffer.ptr);
    zfree(kvArray); zfree(keyArray); zfree(newargv);
    return;

/* On socket errors we try to close the cached socket and try again.
 * It is very common for the cached socket to get closed, if just reopening
 * it works it's a shame to notify the error to the caller. */
    socket_err:
    /* Cleanup we want to perform in both the retry and no retry case.
     * Note: Closing the migrate socket will also force SELECT next time. */
    sdsfree(cmd.io.buffer.ptr);

    /* If the command was rewritten as DEL and there was a socket error,
     * we already closed the socket earlier. While migrateCloseSocket()
     * is idempotent, the host/port arguments are now gone, so don't do it
     * again. */
    if (!argv_rewritten) migrateCloseSocket(c->argv[1],c->argv[2]);
    zfree(newargv);
    newargv = NULL; /* This will get reallocated on retry. */

    /* Retry only if it's not a timeout and we never attempted a retry
     * (or the code jumping here did not set may_retry to zero). */
    if (errno != ETIMEDOUT && may_retry) {
        may_retry = 0;
        goto try_again;
    }

    /* Cleanup we want to do if no retry is attempted. */
    zfree(kvArray); zfree(keyArray);
    addReplyErrorSds(c, sdscatprintf(sdsempty(),
                                     "-IOERR error or timeout %s to target instance",
                                     write_error ? "writing" : "reading"));
    return;
}

/* Cluster node sanity check. Returns C_OK if the node id
 * is valid an C_ERR otherwise. */
int verifyClusterNodeId(const char *name, int length) {
    if (length != CLUSTER_NAMELEN) return C_ERR;
    for (int i = 0; i < length; i++) {
        if (name[i] >= 'a' && name[i] <= 'z') continue;
        if (name[i] >= '0' && name[i] <= '9') continue;
        return C_ERR;
    }
    return C_OK;
}

int isValidAuxChar(int c) {
    /* Reject control characters (0x00-0x1F and 0x7F). */
    if (iscntrl(c)) {
        return 0;
    }
    /* Reject forbidden characters including nodes.conf delimiters and special parsing characters */
    return isalnum(c) || (strchr("!#$%&()*+:;<>?@[]^{|}~,= \"'\\", c) == NULL);
}

int isValidAuxString(char *s, unsigned int length) {
    for (unsigned i = 0; i < length; i++) {
        if (!isValidAuxChar(s[i])) return 0;
    }
    return 1;
}

void clusterCommandMyId(client *c) {
    char *name = clusterNodeGetName(getMyClusterNode());
    if (name) {
        addReplyBulkCBuffer(c,name, CLUSTER_NAMELEN);
    } else {
        addReplyError(c, "No ID yet");
    }
}

char* getMyClusterId(void) {
    return clusterNodeGetName(getMyClusterNode());
}

void clusterCommandMyShardId(client *c) {
    char *sid = clusterNodeGetShardId(getMyClusterNode());
    if (sid) {
        addReplyBulkCBuffer(c,sid, CLUSTER_NAMELEN);
    } else {
        addReplyError(c, "No shard ID yet");
    }
}

/* When a cluster command is called, we need to decide whether to return TLS info or
 * non-TLS info by the client's connection type. However if the command is called by
 * a Lua script or RM_call, there is no connection in the fake client, so we use
 * server.current_client here to get the real client if available. And if it is not
 * available (modules may call commands without a real client), we return the default
 * info, which is determined by server.tls_cluster. */
static int shouldReturnTlsInfo(void) {
    if (server.current_client && server.current_client->conn) {
        return connIsTLS(server.current_client->conn);
    } else {
        return server.tls_cluster;
    }
}

unsigned int countKeysInSlot(unsigned int slot) {
    return kvstoreDictSize(server.db->keys, slot);
}

void removeChannelsInSlot(unsigned int slot) {
    if (countChannelsInSlot(slot) == 0) return;

    pubsubShardUnsubscribeAllChannelsInSlot(slot);
}

/* Get the count of the channels for a given slot. */
unsigned int countChannelsInSlot(unsigned int hashslot) {
    return kvstoreDictSize(server.pubsubshard_channels, hashslot);
}

/* Add detailed information of a node to the output buffer of the given client. */
void addNodeDetailsToShardReply(client *c, clusterNode *node) {

    int reply_count = 0;
    char *hostname;
    void *node_replylen = addReplyDeferredLen(c);

    addReplyBulkCString(c, "id");
    addReplyBulkCBuffer(c, clusterNodeGetName(node), CLUSTER_NAMELEN);
    reply_count++;

    if (clusterNodeTcpPort(node)) {
        addReplyBulkCString(c, "port");
        addReplyLongLong(c, clusterNodeTcpPort(node));
        reply_count++;
    }

    if (clusterNodeTlsPort(node)) {
        addReplyBulkCString(c, "tls-port");
        addReplyLongLong(c, clusterNodeTlsPort(node));
        reply_count++;
    }

    addReplyBulkCString(c, "ip");
    addReplyBulkCString(c, clusterNodeIp(node));
    reply_count++;

    addReplyBulkCString(c, "endpoint");
    addReplyBulkCString(c, clusterNodePreferredEndpoint(node));
    reply_count++;

    hostname = clusterNodeHostname(node);
    if (hostname != NULL && *hostname != '\0') {
        addReplyBulkCString(c, "hostname");
        addReplyBulkCString(c, hostname);
        reply_count++;
    }

    long long node_offset;
    if (clusterNodeIsMyself(node)) {
        node_offset = clusterNodeIsSlave(node) ? replicationGetSlaveOffset() : server.master_repl_offset;
    } else {
        node_offset = clusterNodeReplOffset(node);
    }

    addReplyBulkCString(c, "role");
    addReplyBulkCString(c, clusterNodeIsSlave(node) ? "replica" : "master");
    reply_count++;

    addReplyBulkCString(c, "replication-offset");
    addReplyLongLong(c, node_offset);
    reply_count++;

    addReplyBulkCString(c, "health");
    const char *health_msg = NULL;
    if (clusterNodeIsFailing(node)) {
        health_msg = "fail";
    } else if (clusterNodeIsSlave(node) && node_offset == 0) {
        health_msg = "loading";
    } else {
        health_msg = "online";
    }
    addReplyBulkCString(c, health_msg);
    reply_count++;

    setDeferredMapLen(c, node_replylen, reply_count);
}

static clusterNode *clusterGetMasterFromShard(void *shard_handle) {
    clusterNode *n = NULL;
    void *node_it = clusterShardHandleGetNodeIterator(shard_handle);
    while((n = clusterShardNodeIteratorNext(node_it)) != NULL) {
        if (!clusterNodeIsFailing(n)) {
            break;
        }
    }
    clusterShardNodeIteratorFree(node_it);
    if (!n) return NULL;
    return clusterNodeGetMaster(n);
}

/* Add the shard reply of a single shard based off the given primary node. */
void addShardReplyForClusterShards(client *c, void *shard_handle) {
    serverAssert(clusterGetShardNodeCount(shard_handle) > 0);
    addReplyMapLen(c, 2);
    addReplyBulkCString(c, "slots");

    /* Use slot_info_pairs from the primary only */
    clusterNode *master_node = clusterGetMasterFromShard(shard_handle);

    if (master_node && clusterNodeHasSlotInfo(master_node)) {
        serverAssert((clusterNodeSlotInfoCount(master_node) % 2) == 0);
        addReplyArrayLen(c, clusterNodeSlotInfoCount(master_node));
        for (int i = 0; i < clusterNodeSlotInfoCount(master_node); i++)
            addReplyLongLong(c, (unsigned long)clusterNodeSlotInfoEntry(master_node, i));
    } else {
        /* If no slot info pair is provided, the node owns no slots */
        addReplyArrayLen(c, 0);
    }

    addReplyBulkCString(c, "nodes");
    addReplyArrayLen(c, clusterGetShardNodeCount(shard_handle));
    void *node_it = clusterShardHandleGetNodeIterator(shard_handle);
    for (clusterNode *n = clusterShardNodeIteratorNext(node_it); n != NULL; n = clusterShardNodeIteratorNext(node_it)) {
        addNodeDetailsToShardReply(c, n);
        clusterFreeNodesSlotsInfo(n);
    }
    clusterShardNodeIteratorFree(node_it);
}

/* Add to the output buffer of the given client, an array of slot (start, end)
 * pair owned by the shard, also the primary and set of replica(s) along with
 * information about each node. */
void clusterCommandShards(client *c) {
    addReplyArrayLen(c, clusterGetShardCount());
    /* This call will add slot_info_pairs to all nodes */
    clusterGenNodesSlotsInfo(0);
    dictIterator *shard_it = clusterGetShardIterator();
    for(void *shard_handle = clusterNextShardHandle(shard_it); shard_handle != NULL; shard_handle = clusterNextShardHandle(shard_it)) {
        addShardReplyForClusterShards(c, shard_handle);
    }
    clusterFreeShardIterator(shard_it);
}

void clusterCommandHelp(client *c) {
    const char *help[] = {
            "COUNTKEYSINSLOT <slot>",
            "    Return the number of keys in <slot>.",
            "GETKEYSINSLOT <slot> <count>",
            "    Return key names stored by current node in a slot.",
            "INFO",
            "    Return information about the cluster.",
            "KEYSLOT <key>",
            "    Return the hash slot for <key>.",
            "MYID",
            "    Return the node id.",
            "MYSHARDID",
            "    Return the node's shard id.",
            "NODES",
            "    Return cluster configuration seen by node. Output format:",
            "    <id> <ip:port@bus-port[,hostname]> <flags> <master> <pings> <pongs> <epoch> <link> <slot> ...",
            "REPLICAS <node-id>",
            "    Return <node-id> replicas.",
            "SLOTS",
            "    Return information about slots range mappings. Each range is made of:",
            "    start, end, master and replicas IP addresses, ports and ids",
            "SLOT-STATS",
            "    Return an array of slot usage statistics for slots assigned to the current node.",
            "SHARDS",
            "    Return information about slot range mappings and the nodes associated with them.",
            NULL
    };

    addExtendedReplyHelp(c, help, clusterCommandExtendedHelp());
}

void clusterCommand(client *c) {
    if (server.cluster_enabled == 0) {
        addReplyError(c,"This instance has cluster support disabled");
        return;
    }

    if (c->argc == 2 && !strcasecmp(c->argv[1]->ptr,"help")) {
        clusterCommandHelp(c);
    } else  if (!strcasecmp(c->argv[1]->ptr,"nodes") && c->argc == 2) {
        /* CLUSTER NODES */
        /* Report TLS ports to TLS client, and report non-TLS port to non-TLS client. */
        sds nodes = clusterGenNodesDescription(c, 0, shouldReturnTlsInfo());
        addReplyVerbatim(c,nodes,sdslen(nodes),"txt");
        sdsfree(nodes);
    } else if (!strcasecmp(c->argv[1]->ptr,"myid") && c->argc == 2) {
        /* CLUSTER MYID */
        clusterCommandMyId(c);
    } else if (!strcasecmp(c->argv[1]->ptr,"myshardid") && c->argc == 2) {
        /* CLUSTER MYSHARDID */
        clusterCommandMyShardId(c);
    } else if (!strcasecmp(c->argv[1]->ptr,"slots") && c->argc == 2) {
        /* CLUSTER SLOTS */
        clusterCommandSlots(c);
    } else if (!strcasecmp(c->argv[1]->ptr,"shards") && c->argc == 2) {
        /* CLUSTER SHARDS */
        clusterCommandShards(c);
    } else if (!strcasecmp(c->argv[1]->ptr,"info") && c->argc == 2) {
        /* CLUSTER INFO */

        sds info = genClusterInfoString();

        /* Produce the reply protocol. */
        addReplyVerbatim(c,info,sdslen(info),"txt");
        sdsfree(info);
    } else if (!strcasecmp(c->argv[1]->ptr,"keyslot") && c->argc == 3) {
        /* CLUSTER KEYSLOT <key> */
        sds key = c->argv[2]->ptr;

        addReplyLongLong(c,keyHashSlot(key,sdslen(key)));
    } else if (!strcasecmp(c->argv[1]->ptr,"countkeysinslot") && c->argc == 3) {
        /* CLUSTER COUNTKEYSINSLOT <slot> */
        long long slot;

        if (getLongLongFromObjectOrReply(c,c->argv[2],&slot,NULL) != C_OK)
            return;
        if (slot < 0 || slot >= CLUSTER_SLOTS) {
            addReplyError(c,"Invalid slot");
            return;
        }

        if (!clusterCanAccessKeysInSlot(slot)) {
            addReplyLongLong(c, 0);
            return;
        }
        addReplyLongLong(c,countKeysInSlot(slot));
    } else if (!strcasecmp(c->argv[1]->ptr,"getkeysinslot") && c->argc == 4) {
        /* CLUSTER GETKEYSINSLOT <slot> <count> */
        long long maxkeys, slot;

        if (getLongLongFromObjectOrReply(c,c->argv[2],&slot,NULL) != C_OK)
            return;
        if (getLongLongFromObjectOrReply(c,c->argv[3],&maxkeys,NULL)
            != C_OK)
            return;
        if (slot < 0 || slot >= CLUSTER_SLOTS || maxkeys < 0) {
            addReplyError(c,"Invalid slot or number of keys");
            return;
        }

        if (!clusterCanAccessKeysInSlot(slot)) {
            addReplyArrayLen(c, 0);
            return;
        }

        unsigned int keys_in_slot = countKeysInSlot(slot);
        unsigned int numkeys = maxkeys > keys_in_slot ? keys_in_slot : maxkeys;
        addReplyArrayLen(c,numkeys);
        kvstoreDictIterator kvs_di;
        dictEntry *de = NULL;
        kvstoreInitDictIterator(&kvs_di, server.db->keys, slot);
        for (unsigned int i = 0; i < numkeys; i++) {
            de = kvstoreDictIteratorNext(&kvs_di);
            serverAssert(de != NULL);
            sds sdskey = kvobjGetKey(dictGetKV(de));
            addReplyBulkCBuffer(c, sdskey, sdslen(sdskey));
        }
        kvstoreResetDictIterator(&kvs_di);
    } else if ((!strcasecmp(c->argv[1]->ptr,"slaves") ||
                !strcasecmp(c->argv[1]->ptr,"replicas")) && c->argc == 3) {
        /* CLUSTER SLAVES <NODE ID> */
        /* CLUSTER REPLICAS <NODE ID> */
        clusterNode *n = clusterLookupNode(c->argv[2]->ptr, sdslen(c->argv[2]->ptr));
        int j;

        /* Lookup the specified node in our table. */
        if (!n) {
            addReplyErrorFormat(c,"Unknown node %s", (char*)c->argv[2]->ptr);
            return;
        }

        if (clusterNodeIsSlave(n)) {
            addReplyError(c,"The specified node is not a master");
            return;
        }

        /* Report TLS ports to TLS client, and report non-TLS port to non-TLS client. */
        addReplyArrayLen(c, clusterNodeNumSlaves(n));
        for (j = 0; j < clusterNodeNumSlaves(n); j++) {
            sds ni = clusterGenNodeDescription(c, clusterNodeGetSlave(n, j), shouldReturnTlsInfo());
            addReplyBulkCString(c,ni);
            sdsfree(ni);
        }
    } else if (!strcasecmp(c->argv[1]->ptr, "migration")) {
        clusterMigrationCommand(c);
    } else if (!strcasecmp(c->argv[1]->ptr,"syncslots") && c->argc >= 3) {
        clusterSyncSlotsCommand(c);
    } else if(!clusterCommandSpecial(c)) {
        addReplySubcommandSyntaxError(c);
        return;
    }
}

/* Extract slot number from keys in a keys_result structure and return to caller.
 * Returns:
 *   - The slot number if all keys belong to the same slot
 *   - INVALID_CLUSTER_SLOT if there are no keys or cluster is disabled
 *   - CLUSTER_CROSSSLOT if keys belong to different slots (cross-slot error) */
int extractSlotFromKeysResult(robj **argv, getKeysResult *keys_result) {
    if (keys_result->numkeys == 0 || !server.cluster_enabled)
        return INVALID_CLUSTER_SLOT;

    int first_slot = INVALID_CLUSTER_SLOT;
    for (int j = 0; j < keys_result->numkeys; j++) {
        robj *this_key = argv[keys_result->keys[j].pos];
        int this_slot = (int)keyHashSlot((char*)this_key->ptr, sdslen(this_key->ptr));

        if (first_slot == INVALID_CLUSTER_SLOT)
            first_slot = this_slot;
        else if (first_slot != this_slot) {
            return CLUSTER_CROSSSLOT;
        }
    }
    return first_slot;
}

/* Return the pointer to the cluster node that is able to serve the command.
 * For the function to succeed the command should only target either:
 *
 * 1) A single key (even multiple times like RPOPLPUSH mylist mylist).
 * 2) Multiple keys in the same hash slot, while the slot is stable (no
 *    resharding in progress).
 *
 * On success the function returns the node that is able to serve the request.
 * If the node is not 'myself' a redirection must be performed. The kind of
 * redirection is specified setting the integer passed by reference
 * 'error_code', which will be set to CLUSTER_REDIR_ASK or
 * CLUSTER_REDIR_MOVED.
 *
 * When the node is 'myself' 'error_code' is set to CLUSTER_REDIR_NONE.
 *
 * If the command fails NULL is returned, and the reason of the failure is
 * provided via 'error_code', which will be set to:
 *
 * CLUSTER_REDIR_CROSS_SLOT if the request contains multiple keys that
 * don't belong to the same hash slot.
 *
 * CLUSTER_REDIR_UNSTABLE if the request contains multiple keys
 * belonging to the same slot, but the slot is not stable (in migration or
 * importing state, likely because a resharding is in progress).
 *
 * CLUSTER_REDIR_DOWN_UNBOUND if the request addresses a slot which is
 * not bound to any node. In this case the cluster global state should be
 * already "down" but it is fragile to rely on the update of the global state,
 * so we also handle it here.
 *
 * CLUSTER_REDIR_TRIMMING if the request addresses a slot that is being trimmed.
 *
 * CLUSTER_REDIR_DOWN_STATE and CLUSTER_REDIR_DOWN_RO_STATE if the cluster is
 * down but the user attempts to execute a command that addresses one or more keys. */
clusterNode *getNodeByQuery(client *c, struct redisCommand *cmd, robj **argv, int argc, int *hashslot,
    getKeysResult *keys_result, uint8_t read_error, uint64_t cmd_flags, int *error_code)
{
    clusterNode *myself = getMyClusterNode();
    clusterNode *n = NULL;
    robj *firstkey = NULL;
    int multiple_keys = 0;
    multiState *ms, _ms;
    pendingCommand mc;
    pendingCommand *mcp = &mc;
    int i, slot = 0, migrating_slot = 0, importing_slot = 0, missing_keys = 0,
            existing_keys = 0;
    int pubsubshard_included = 0; /* Flag to indicate if a pubsub shard cmd is included. */

    /* Allow any key to be set if a module disabled cluster redirections. */
    if (server.cluster_module_flags & CLUSTER_MODULE_FLAG_NO_REDIRECTION)
        return myself;

    /* Set error code optimistically for the base case. */
    if (error_code) *error_code = CLUSTER_REDIR_NONE;

    /* Modules can turn off Redis Cluster redirection: this is useful
     * when writing a module that implements a completely different
     * distributed system. */

    /* We handle all the cases as if they were EXEC commands, so we have
     * a common code path for everything */
    if (cmd->proc == execCommand) {
        /* If CLIENT_MULTI flag is not set EXEC is just going to return an
         * error. */
        if (!(c->flags & CLIENT_MULTI)) return myself;
        ms = &c->mstate;
    } else {
        /* In order to have a single codepath create a fake Multi State
         * structure if the client is not in MULTI/EXEC state, this way
         * we have a single codepath below. */
        ms = &_ms;
        _ms.commands = &mcp;
        _ms.count = 1;

        /* Properly initialize the fake pendingCommand */
        initPendingCommand(&mc);
        mc.argv = argv;
        mc.argc = argc;
        mc.cmd = cmd;
        mc.slot = hashslot ? *hashslot : INVALID_CLUSTER_SLOT;
        mc.read_error = read_error;
        if (keys_result) {
            mc.keys_result = *keys_result;
            mc.flags |= PENDING_CMD_KEYS_RESULT_VALID;
        }
    }

    /* Check that all the keys are in the same hash slot, and obtain this
     * slot and the node associated. */
    for (i = 0; i < ms->count; i++) {
        struct redisCommand *mcmd;
        robj **margv;
        int margc, j;
        keyReference *keyindex;

        pendingCommand *pcmd = ms->commands[i];

        mcmd = pcmd->cmd;
        margc = pcmd->argc;
        margv = pcmd->argv;

        /* Only valid for sharded pubsub as regular pubsub can operate on any node and bypasses this layer. */
        if (!pubsubshard_included &&
            doesCommandHaveChannelsWithFlags(mcmd, CMD_CHANNEL_PUBLISH | CMD_CHANNEL_SUBSCRIBE))
        {
            pubsubshard_included = 1;
        }

        /* If we have a cached keys result from preprocessCommand(), use it.
         * Otherwise, extract keys result. */
        int use_cache_keys_result = pcmd->flags & PENDING_CMD_KEYS_RESULT_VALID;
        getKeysResult result = GETKEYS_RESULT_INIT;
        if (use_cache_keys_result)
            result = pcmd->keys_result;
        else
            getKeysFromCommand(mcmd,margv,margc,&result);
        keyindex = result.keys;

        for (j = 0; j < result.numkeys; j++) {
            /* The command has keys and was checked for cross-slot between its keys in preprocessCommand() */
            if (pcmd->read_error == CLIENT_READ_CROSS_SLOT) {
                /* Error: multiple keys from different slots. */
                if (!use_cache_keys_result) getKeysFreeResult(&result);
                if (error_code)
                    *error_code = CLUSTER_REDIR_CROSS_SLOT;
                return NULL;
            }

            robj *thiskey = margv[keyindex[j].pos];
            int thisslot = pcmd->slot;
            if (thisslot == INVALID_CLUSTER_SLOT)
                thisslot = keyHashSlot((char*)thiskey->ptr, sdslen(thiskey->ptr));

            if (firstkey == NULL) {
                /* This is the first key we see. Check what is the slot
                 * and node. */
                firstkey = thiskey;
                slot = thisslot;
                n = getNodeBySlot(slot);

                /* Error: If a slot is not served, we are in "cluster down"
                 * state. However the state is yet to be updated, so this was
                 * not trapped earlier in processCommand(). Report the same
                 * error to the client. */
                if (n == NULL) {
                    if (!use_cache_keys_result) getKeysFreeResult(&result);
                    if (error_code)
                        *error_code = CLUSTER_REDIR_DOWN_UNBOUND;
                    return NULL;
                }

                /* If we are migrating or importing this slot, we need to check
                 * if we have all the keys in the request (the only way we
                 * can safely serve the request, otherwise we return a TRYAGAIN
                 * error). To do so we set the importing/migrating state and
                 * increment a counter for every missing key. */
                if (n == myself &&
                    getMigratingSlotDest(slot) != NULL)
                {
                    migrating_slot = 1;
                } else if (getImportingSlotSource(slot) != NULL) {
                    importing_slot = 1;
                }
            } else {
                /* If it is not the first key/channel, make sure it is exactly
                 * the same key/channel as the first we saw. */
                if (slot != thisslot) {
                    /* Error: multiple keys from different slots. */
                    if (!use_cache_keys_result) getKeysFreeResult(&result);
                    if (error_code)
                        *error_code = CLUSTER_REDIR_CROSS_SLOT;
                    return NULL;
                }
                if (importing_slot && !multiple_keys && !equalStringObjects(firstkey,thiskey)) {
                    /* Flag this request as one with multiple different
                     * keys/channels when the slot is in importing state. */
                    multiple_keys = 1;
                }
            }

            /* Migrating / Importing slot? Count keys we don't have.
             * If it is pubsubshard command, it isn't required to check
             * the channel being present or not in the node during the
             * slot migration, the channel will be served from the source
             * node until the migration completes with CLUSTER SETSLOT <slot>
             * NODE <node-id>. */
            int flags = LOOKUP_NOTOUCH | LOOKUP_NOSTATS | LOOKUP_NONOTIFY | LOOKUP_NOEXPIRE;
            if ((migrating_slot || importing_slot) && !pubsubshard_included)
            {
                if (lookupKeyReadWithFlags(&server.db[0], thiskey, flags) == NULL) missing_keys++;
                else existing_keys++;
            }
        }
        if (!use_cache_keys_result) getKeysFreeResult(&result);
    }

    /* No key at all in command? then we can serve the request
     * without redirections or errors in all the cases. */
    if (n == NULL) return myself;

    /* Cluster is globally down but we got keys? We only serve the request
     * if it is a read command and when allow_reads_when_down is enabled. */
    if (!isClusterHealthy()) {
        if (pubsubshard_included) {
            if (!server.cluster_allow_pubsubshard_when_down) {
                if (error_code) *error_code = CLUSTER_REDIR_DOWN_STATE;
                return NULL;
            }
        } else if (!server.cluster_allow_reads_when_down) {
            /* The cluster is configured to block commands when the
             * cluster is down. */
            if (error_code) *error_code = CLUSTER_REDIR_DOWN_STATE;
            return NULL;
        } else if (cmd_flags & CMD_WRITE) {
            /* The cluster is configured to allow read only commands */
            if (error_code) *error_code = CLUSTER_REDIR_DOWN_RO_STATE;
            return NULL;
        } else {
            /* Fall through and allow the command to be executed:
             * this happens when server.cluster_allow_reads_when_down is
             * true and the command is not a write command */
        }
    }

    /* Return the hashslot by reference. */
    if (hashslot) *hashslot = slot;

    /* MIGRATE always works in the context of the local node if the slot
     * is open (migrating or importing state). We need to be able to freely
     * move keys among instances in this case. */
    if ((migrating_slot || importing_slot) && cmd->proc == migrateCommand)
        return myself;

    /* If we don't have all the keys and we are migrating the slot, send
     * an ASK redirection or TRYAGAIN. */
    if (migrating_slot && missing_keys) {
        /* If we have keys but we don't have all keys, we return TRYAGAIN */
        if (existing_keys) {
            if (error_code) *error_code = CLUSTER_REDIR_UNSTABLE;
            return NULL;
        } else {
            if (error_code) *error_code = CLUSTER_REDIR_ASK;
            return getMigratingSlotDest(slot);
        }
    }

    /* If we are receiving the slot, and the client correctly flagged the
     * request as "ASKING", we can serve the request. However if the request
     * involves multiple keys and we don't have them all, the only option is
     * to send a TRYAGAIN error. */
    if (importing_slot &&
        (c->flags & CLIENT_ASKING || cmd_flags & CMD_ASKING))
    {
        if (multiple_keys && missing_keys) {
            if (error_code) *error_code = CLUSTER_REDIR_UNSTABLE;
            return NULL;
        } else {
            return myself;
        }
    }

    /* Handle the read-only client case reading from a slave: if this
     * node is a slave and the request is about a hash slot our master
     * is serving, we can reply without redirection. */
    int is_write_command = (cmd_flags & CMD_WRITE) ||
                           (c->cmd->proc == execCommand && (c->mstate.cmd_flags & CMD_WRITE));
    if (((c->flags & CLIENT_READONLY) || pubsubshard_included) &&
        !is_write_command &&
        clusterNodeIsSlave(myself) &&
        clusterNodeGetSlaveof(myself) == n)
    {
        return myself;
    }

    /* If this node is responsible for the slot and is currently trimming it,
     * SFLUSH may have triggered active trimming and it could still be in progress.
     * Here we reject any write commands as no writes should be accepted for
     * trimming slots while active trimming is in progress. */
    if (n == myself && is_write_command && isSlotInTrimJob(slot)) {
        if (error_code) *error_code = CLUSTER_REDIR_TRIMMING;
        return NULL;
    }

    /* Base case: just return the right node. However, if this node is not
     * myself, set error_code to MOVED since we need to issue a redirection. */
    if (n != myself && error_code) *error_code = CLUSTER_REDIR_MOVED;
    return n;
}

/* Send the client the right redirection code, according to error_code
 * that should be set to one of CLUSTER_REDIR_* macros.
 *
 * If CLUSTER_REDIR_ASK or CLUSTER_REDIR_MOVED error codes
 * are used, then the node 'n' should not be NULL, but should be the
 * node we want to mention in the redirection. Moreover hashslot should
 * be set to the hash slot that caused the redirection. */
void clusterRedirectClient(client *c, clusterNode *n, int hashslot, int error_code) {
    if (error_code == CLUSTER_REDIR_CROSS_SLOT) {
        addReplyError(c,"-CROSSSLOT Keys in request don't hash to the same slot");
    } else if (error_code == CLUSTER_REDIR_UNSTABLE) {
        /* The request spawns multiple keys in the same slot,
         * but the slot is not "stable" currently as there is
         * a migration or import in progress. */
        addReplyError(c,"-TRYAGAIN Multiple keys request during rehashing of slot");
    } else if (error_code == CLUSTER_REDIR_DOWN_STATE) {
        addReplyError(c,"-CLUSTERDOWN The cluster is down");
    } else if (error_code == CLUSTER_REDIR_DOWN_RO_STATE) {
        addReplyError(c,"-CLUSTERDOWN The cluster is down and only accepts read commands");
    } else if (error_code == CLUSTER_REDIR_DOWN_UNBOUND) {
        addReplyError(c,"-CLUSTERDOWN Hash slot not served");
    } else if (error_code == CLUSTER_REDIR_MOVED ||
               error_code == CLUSTER_REDIR_ASK)
    {
        /* Report TLS ports to TLS client, and report non-TLS port to non-TLS client. */
        int port = clusterNodeClientPort(n, shouldReturnTlsInfo());
        addReplyErrorSds(c,sdscatprintf(sdsempty(),
                                        "-%s %d %s:%d",
                                        (error_code == CLUSTER_REDIR_ASK) ? "ASK" : "MOVED",
                                        hashslot, clusterNodePreferredEndpoint(n), port));
    } else if (error_code == CLUSTER_REDIR_TRIMMING) {
        addReplyError(c,"-TRYAGAIN Slot is being trimmed");
    } else {
        serverPanic("getNodeByQuery() unknown error.");
    }
}

/* This function is called by the function processing clients incrementally
 * to detect timeouts, in order to handle the following case:
 *
 * 1) A client blocks with BLPOP or similar blocking operation.
 * 2) The master migrates the hash slot elsewhere or turns into a slave.
 * 3) The client may remain blocked forever (or up to the max timeout time)
 *    waiting for a key change that will never happen.
 *
 * If the client is found to be blocked into a hash slot this node no
 * longer handles, the client is sent a redirection error, and the function
 * returns 1. Otherwise 0 is returned and no operation is performed. */
int clusterRedirectBlockedClientIfNeeded(client *c) {
    clusterNode *myself = getMyClusterNode();
    if (c->flags & CLIENT_BLOCKED &&
        (c->bstate.btype == BLOCKED_LIST ||
         c->bstate.btype == BLOCKED_LIST_NONEMPTY ||
         c->bstate.btype == BLOCKED_ZSET ||
         c->bstate.btype == BLOCKED_STREAM ||
         c->bstate.btype == BLOCKED_MODULE))
    {
        dictEntry *de;
        dictIterator di;

        /* If the cluster is down, unblock the client with the right error.
         * If the cluster is configured to allow reads on cluster down, we
         * still want to emit this error since a write will be required
         * to unblock them which may never come.  */
        if (!isClusterHealthy()) {
            clusterRedirectClient(c,NULL,0,CLUSTER_REDIR_DOWN_STATE);
            return 1;
        }

        /* If the client is blocked on module, but not on a specific key,
         * don't unblock it (except for the CLUSTER_FAIL case above). */
        if (c->bstate.btype == BLOCKED_MODULE && !moduleClientIsBlockedOnKeys(c))
            return 0;

        /* All keys must belong to the same slot, so check first key only. */
        dictInitIterator(&di, c->bstate.keys);
        if ((de = dictNext(&di)) != NULL) {
            robj *key = dictGetKey(de);
            int slot = keyHashSlot((char*)key->ptr, sdslen(key->ptr));
            clusterNode *node = getNodeBySlot(slot);

            /* if the client is read-only and attempting to access key that our
             * replica can handle, allow it. */
            if ((c->flags & CLIENT_READONLY) &&
                !(c->lastcmd->flags & CMD_WRITE) &&
                clusterNodeIsSlave(myself) && clusterNodeGetSlaveof(myself) == node)
            {
                node = myself;
            }

            /* We send an error and unblock the client if:
             * 1) The slot is unassigned, emitting a cluster down error.
             * 2) The slot is not handled by this node, nor being imported. */
            if (node != myself && getImportingSlotSource(slot) == NULL)
            {
                if (node == NULL) {
                    clusterRedirectClient(c,NULL,0,
                                          CLUSTER_REDIR_DOWN_UNBOUND);
                } else {
                    clusterRedirectClient(c,node,slot,
                                          CLUSTER_REDIR_MOVED);
                }
                dictResetIterator(&di);
                return 1;
            }
        }
        dictResetIterator(&di);
    }
    return 0;
}

/* Returns an indication if the replica node is fully available
 * and should be listed in CLUSTER SLOTS response.
 * Returns 1 for available nodes, 0 for nodes that have
 * not finished their initial sync, in failed state, or are
 * otherwise considered not available to serve read commands. */
static int isReplicaAvailable(clusterNode *node) {
    if (clusterNodeIsFailing(node)) {
        return 0;
    }
    long long repl_offset = clusterNodeReplOffset(node);
    if (clusterNodeIsMyself(node)) {
        /* Nodes do not update their own information
         * in the cluster node list. */
        repl_offset = replicationGetSlaveOffset();
    }
    return (repl_offset != 0);
}

void addNodeToNodeReply(client *c, clusterNode *node) {
    char* hostname = clusterNodeHostname(node);
    addReplyArrayLen(c, 4);
    if (server.cluster_preferred_endpoint_type == CLUSTER_ENDPOINT_TYPE_IP) {
        addReplyBulkCString(c, clusterNodeIp(node));
    } else if (server.cluster_preferred_endpoint_type == CLUSTER_ENDPOINT_TYPE_HOSTNAME) {
        if (hostname != NULL && hostname[0] != '\0') {
            addReplyBulkCString(c, hostname);
        } else {
            addReplyBulkCString(c, "?");
        }
    } else if (server.cluster_preferred_endpoint_type == CLUSTER_ENDPOINT_TYPE_UNKNOWN_ENDPOINT) {
        addReplyNull(c);
    } else {
        serverPanic("Unrecognized preferred endpoint type");
    }

    /* Report TLS ports to TLS client, and report non-TLS port to non-TLS client. */
    addReplyLongLong(c, clusterNodeClientPort(node, shouldReturnTlsInfo()));
    addReplyBulkCBuffer(c, clusterNodeGetName(node), CLUSTER_NAMELEN);

    /* Add the additional endpoint information, this is all the known networking information
     * that is not the preferred endpoint. Note the logic is evaluated twice so we can
     * correctly report the number of additional network arguments without using a deferred
     * map, an assertion is made at the end to check we set the right length. */
    int length = 0;
    if (server.cluster_preferred_endpoint_type != CLUSTER_ENDPOINT_TYPE_IP) {
        length++;
    }
    if (server.cluster_preferred_endpoint_type != CLUSTER_ENDPOINT_TYPE_HOSTNAME
        && hostname != NULL && hostname[0] != '\0')
    {
        length++;
    }
    addReplyMapLen(c, length);

    if (server.cluster_preferred_endpoint_type != CLUSTER_ENDPOINT_TYPE_IP) {
        addReplyBulkCString(c, "ip");
        addReplyBulkCString(c, clusterNodeIp(node));
        length--;
    }
    if (server.cluster_preferred_endpoint_type != CLUSTER_ENDPOINT_TYPE_HOSTNAME
        && hostname != NULL && hostname[0] != '\0')
    {
        addReplyBulkCString(c, "hostname");
        addReplyBulkCString(c, hostname);
        length--;
    }
    serverAssert(length == 0);
}

void addNodeReplyForClusterSlot(client *c, clusterNode *node, int start_slot, int end_slot) {
    int i, nested_elements = 3; /* slots (2) + master addr (1) */
    for (i = 0; i < clusterNodeNumSlaves(node); i++) {
        if (!isReplicaAvailable(clusterNodeGetSlave(node, i))) continue;
        nested_elements++;
    }
    addReplyArrayLen(c, nested_elements);
    addReplyLongLong(c, start_slot);
    addReplyLongLong(c, end_slot);
    addNodeToNodeReply(c, node);

    /* Remaining nodes in reply are replicas for slot range */
    for (i = 0; i < clusterNodeNumSlaves(node); i++) {
        /* This loop is copy/pasted from clusterGenNodeDescription()
         * with modifications for per-slot node aggregation. */
        if (!isReplicaAvailable(clusterNodeGetSlave(node, i))) continue;
        addNodeToNodeReply(c, clusterNodeGetSlave(node, i));
        nested_elements--;
    }
    serverAssert(nested_elements == 3); /* Original 3 elements */
}

void clusterCommandSlots(client * c) {
    /* Format: 1) 1) start slot
     *            2) end slot
     *            3) 1) master IP
     *               2) master port
     *               3) node ID
     *            4) 1) replica IP
     *               2) replica port
     *               3) node ID
     *           ... continued until done
     */
    clusterNode *n = NULL;
    int num_masters = 0, start = -1;
    void *slot_replylen = addReplyDeferredLen(c);

    for (int i = 0; i <= CLUSTER_SLOTS; i++) {
        /* Find start node and slot id. */
        if (n == NULL) {
            if (i == CLUSTER_SLOTS) break;
            n = getNodeBySlot(i);
            start = i;
            continue;
        }

        /* Add cluster slots info when occur different node with start
         * or end of slot. */
        if (i == CLUSTER_SLOTS || n != getNodeBySlot(i)) {
            addNodeReplyForClusterSlot(c, n, start, i-1);
            num_masters++;
            if (i == CLUSTER_SLOTS) break;
            n = getNodeBySlot(i);
            start = i;
        }
    }
    setDeferredArrayLen(c, slot_replylen, num_masters);
}

/* -----------------------------------------------------------------------------
 * Cluster functions related to serving / redirecting clients
 * -------------------------------------------------------------------------- */

/* The ASKING command is required after a -ASK redirection.
 * The client should issue ASKING before to actually send the command to
 * the target instance. See the Redis Cluster specification for more
 * information. */
void askingCommand(client *c) {
    if (server.cluster_enabled == 0) {
        addReplyError(c,"This instance has cluster support disabled");
        return;
    }
    c->flags |= CLIENT_ASKING;
    addReply(c,shared.ok);
}

/* The READONLY command is used by clients to enter the read-only mode.
 * In this mode slaves will not redirect clients as long as clients access
 * with read-only commands to keys that are served by the slave's master. */
void readonlyCommand(client *c) {
    if (server.cluster_enabled == 0) {
        addReplyError(c,"This instance has cluster support disabled");
        return;
    }
    c->flags |= CLIENT_READONLY;
    addReply(c,shared.ok);
}

/* Remove all the keys in the specified hash slot.
 * The number of removed items is returned. */
unsigned int clusterDelKeysInSlot(unsigned int hashslot, int by_command) {
    unsigned int j = 0;

    if (!kvstoreDictSize(server.db->keys, (int) hashslot))
        return 0;

    kvstoreDictIterator kvs_di;
    dictEntry *de = NULL;
    kvstoreInitDictSafeIterator(&kvs_di, server.db->keys, (int) hashslot);
    while((de = kvstoreDictIteratorNext(&kvs_di)) != NULL) {
        enterExecutionUnit(1, 0);
        sds sdskey = kvobjGetKey(dictGetKV(de));
        robj *key = createStringObject(sdskey, sdslen(sdskey));
        dbDelete(&server.db[0], key);

        keyModified(NULL, &server.db[0], key, NULL, 1);
        if (by_command) {
            /* Keys are deleted by a command (trimslots), we need to notify the
             * keyspace event. Though, we don't need to propagate the DEL
             * command, as the command (trimslots) will be propagated. */
            notifyKeyspaceEvent(NOTIFY_GENERIC, "del", key, server.db[0].id);
        } else {
            /* Propagate the DEL command */
            propagateDeletion(&server.db[0], key, server.lazyfree_lazy_server_del);
            /* The keys are not actually logically deleted from the database,
             * just moved to another node. The modules needs to know that these
             * keys are no longer available locally, so just send the keyspace
             * notification to the modules, but not to clients. */
            moduleNotifyKeyspaceEvent(NOTIFY_GENERIC, "del", key, server.db[0].id, NULL, 0);
        }
        exitExecutionUnit();
        postExecutionUnitOperations();
        decrRefCount(key);
        j++;
        server.dirty++;
    }
    kvstoreResetDictIterator(&kvs_di);
    return j;
}

/* Delete the keys in the slot ranges. Returns the number of deleted items */
unsigned int clusterDelKeysInSlotRangeArray(slotRangeArray *slots, int by_command) {
    unsigned int j = 0;
    for (int i = 0; i < slots->num_ranges; i++) {
        for (int slot = slots->ranges[i].start; slot <= slots->ranges[i].end; slot++) {
            j += clusterDelKeysInSlot(slot, by_command);
        }
    }
    return j;
}

int clusterIsMySlot(int slot) {
    return getMyClusterNode() == getNodeBySlot(slot);
}

void replySlotsFlush(client *c, slotRangeArray *slots) {
    addReplyArrayLen(c, slots->num_ranges);
    for (int i = 0 ; i < slots->num_ranges ; i++) {
        addReplyArrayLen(c, 2);
        addReplyLongLong(c, slots->ranges[i].start);
        addReplyLongLong(c, slots->ranges[i].end);
    }
}

/* Normalizes (sorts and merges adjacent ranges), checks that slot ranges are
 * well-formed and non-overlapping. */
int slotRangeArrayNormalizeAndValidate(slotRangeArray *slots, sds *err) {
    unsigned char used_slots[CLUSTER_SLOTS] = {0};

    if (slots->num_ranges <= 0 || slots->num_ranges >= CLUSTER_SLOTS) {
        *err = sdscatprintf(sdsempty(), "invalid number of slot ranges: %d", slots->num_ranges);
        return C_ERR;
    }

    /* Sort and merge adjacent slot ranges. */
    slotRangeArraySortAndMerge(slots);

    for (int i = 0; i < slots->num_ranges; i++) {
        if (slots->ranges[i].start >= CLUSTER_SLOTS ||
            slots->ranges[i].end >= CLUSTER_SLOTS)
        {
            *err = sdscatprintf(sdsempty(), "slot range is out of range: %d-%d",
                                slots->ranges[i].start, slots->ranges[i].end);
            return C_ERR;
        }

        if (slots->ranges[i].start > slots->ranges[i].end) {
            *err = sdscatprintf(sdsempty(), "start slot number %d is greater than end slot number %d",
                                slots->ranges[i].start, slots->ranges[i].end);
            return C_ERR;
        }

        for (int j = slots->ranges[i].start; j <= slots->ranges[i].end; j++) {
            if (used_slots[j]) {
                *err = sdscatprintf(sdsempty(), "Slot %d specified multiple times", j);
                return C_ERR;
            }
            used_slots[j]++;
        }
    }
    return C_OK;
}

/* Create a slot range array with the specified number of ranges. */
slotRangeArray *slotRangeArrayCreate(int num_ranges) {
    slotRangeArray *slots = zcalloc(sizeof(slotRangeArray) + num_ranges * sizeof(slotRange));
    slots->num_ranges = num_ranges;
    return slots;
}

/* Duplicate the slot range array. */
slotRangeArray *slotRangeArrayDup(slotRangeArray *slots) {
    slotRangeArray *dup = slotRangeArrayCreate(slots->num_ranges);
    memcpy(dup->ranges, slots->ranges, sizeof(slotRange) * slots->num_ranges);
    return dup;
}

/* Set the slot range at the specified index. */
void slotRangeArraySet(slotRangeArray *slots, int idx, int start, int end) {
    slots->ranges[idx].start = start;
    slots->ranges[idx].end = end;
}

/* Create a slot range string in the format of: "1000-2000 3000-4000 ..." */
sds slotRangeArrayToString(slotRangeArray *slots) {
    sds s = sdsempty();
    if (slots == NULL || slots->num_ranges == 0) return s;

    for (int i = 0; i < slots->num_ranges; i++) {
        slotRange *sr = &slots->ranges[i];
        s = sdscatprintf(s, "%d-%d ", sr->start, sr->end);
    }
    sdssetlen(s, sdslen(s) - 1);
    s[sdslen(s)] = '\0';

    return s;
}

/* Parse a slot range string in the format "1000-2000 3000-4000 ..." into a slotRangeArray.
 * Returns a new slotRangeArray on success, NULL on failure. */
slotRangeArray *slotRangeArrayFromString(sds data) {
    int num_ranges;
    long long start, end;
    slotRangeArray *slots = NULL;
    if (!data || sdslen(data) == 0) return NULL;

    sds *parts = sdssplitlen(data, sdslen(data), " ", 1, &num_ranges);
    if (num_ranges <= 0) goto err;

    slots = slotRangeArrayCreate(num_ranges);

    /* Parse each slot range */
    for (int i = 0; i < num_ranges; i++) {
        char *dash = strchr(parts[i], '-');
        if (!dash) goto err;

        if (string2ll(parts[i], dash - parts[i], &start) == 0 ||
            string2ll(dash + 1, sdslen(parts[i]) - (dash - parts[i]) - 1, &end) == 0)
            goto err;
        slotRangeArraySet(slots, i, start, end);
    }

    /* Validate all ranges */
    sds err_msg = NULL;
    if (slotRangeArrayNormalizeAndValidate(slots, &err_msg) != C_OK) {
        if (err_msg) sdsfree(err_msg);
        goto err;
    }
    sdsfreesplitres(parts, num_ranges);
    return slots;

err:
    if (slots) slotRangeArrayFree(slots);
    sdsfreesplitres(parts, num_ranges);
    return NULL;
}

static int compareSlotRange(const void *a, const void *b) {
    const slotRange *sa = a;
    const slotRange *sb = b;
    if (sa->start < sb->start) return -1;
    if (sa->start > sb->start) return 1;
    return 0;
}

/* Sort slot ranges by start slot and merge adjacent ranges.
 * Adjacent means: prev.end + 1 == next.start.
 * e.g. 1000-2000 2001-3000 0-100  =>  0-100 1000-3000
 *
 * Note: Overlapping ranges are not merged.*/
void slotRangeArraySortAndMerge(slotRangeArray *slots) {
    if (!slots || slots->num_ranges <= 1) return;

    qsort(slots->ranges, slots->num_ranges, sizeof(slotRange), compareSlotRange);

    int idx = 0;
    for (int i = 1; i < slots->num_ranges; i++) {
        if (slots->ranges[idx].end + 1 == slots->ranges[i].start)
            slots->ranges[idx].end = slots->ranges[i].end;
        else
            slots->ranges[++idx] = slots->ranges[i];
    }
    slots->num_ranges = idx + 1;
}

/* Compare two slot range arrays, return 1 if equal, 0 otherwise */
int slotRangeArrayIsEqual(slotRangeArray *slots1, slotRangeArray *slots2) {
    slotRangeArraySortAndMerge(slots1);
    slotRangeArraySortAndMerge(slots2);

    if (slots1->num_ranges != slots2->num_ranges) return 0;

    for (int i = 0; i < slots1->num_ranges; i++) {
        if (slots1->ranges[i].start != slots2->ranges[i].start ||
            slots1->ranges[i].end != slots2->ranges[i].end) {
            return 0;
        }
    }
    return 1;
}

/* Add a slot to the slot range array.
 * Usage:
 *     slotRangeArray *slots = NULL
 *     slots = slotRangeArrayAppend(slots, 1000);
 *     slots = slotRangeArrayAppend(slots, 1001);
 *     slots = slotRangeArrayAppend(slots, 1003);
 *     slots = slotRangeArrayAppend(slots, 1004);
 *     slots = slotRangeArrayAppend(slots, 1005);
 *
 *     Result: 1000-1001, 1003-1005
 *     Note: `slot` must be greater than the previous slot.
 * */
slotRangeArray *slotRangeArrayAppend(slotRangeArray *slots, int slot) {
    if (slots == NULL) {
        slots = slotRangeArrayCreate(4);
        slots->ranges[0].start = slot;
        slots->ranges[0].end = slot;
        slots->num_ranges = 1;
        return slots;
    }

    serverAssert(slots->num_ranges >= 0 && slots->num_ranges <= CLUSTER_SLOTS);
    serverAssert(slot > slots->ranges[slots->num_ranges - 1].end);

    /* Check if we can extend the last range */
    slotRange *last = &slots->ranges[slots->num_ranges - 1];
    if (slot == last->end + 1) {
        last->end = slot;
        return slots;
    }

    /* Calculate current capacity and reallocate if needed */
    int cap = (int) ((zmalloc_size(slots) - sizeof(slotRangeArray)) / sizeof(slotRange));
    if (slots->num_ranges >= cap)
        slots = zrealloc(slots, sizeof(slotRangeArray) + sizeof(slotRange) * cap * 2);

    /* Add new single-slot range */
    slots->ranges[slots->num_ranges].start = slot;
    slots->ranges[slots->num_ranges].end = slot;
    slots->num_ranges++;

    return slots;
}

/* Returns 1 if the slot range array contains the given slot, 0 otherwise. */
int slotRangeArrayContains(slotRangeArray *slots, unsigned int slot) {
    for (int i = 0; i < slots->num_ranges; i++)
        if (slots->ranges[i].start <= slot && slots->ranges[i].end >= slot)
            return 1;
    return 0;
}

/* Free the slot range array. */
void slotRangeArrayFree(slotRangeArray *slots) {
    zfree(slots);
}

/* Generic version of slotRangeArrayFree(). */
void slotRangeArrayFreeGeneric(void *slots) {
    slotRangeArrayFree(slots);
}

/* Returns the number of keys in the given slot ranges. */
unsigned long long getKeyCountInSlotRangeArray(slotRangeArray *slots) {
    if (!slots) return 0;

    unsigned long long key_count = 0;
    for (int i = 0; i < slots->num_ranges; i++) {
        for (int j = slots->ranges[i].start; j <= slots->ranges[i].end; j++) {
            key_count += countKeysInSlot(j);
        }
    }
    return key_count;
}

/* Slot range array iterator */
slotRangeArrayIter *slotRangeArrayGetIterator(slotRangeArray *slots) {
    slotRangeArrayIter *it = zmalloc(sizeof(*it));
    it->slots = slots;
    it->range_index = 0;
    it->cur_slot = slots->num_ranges > 0 ? slots->ranges[0].start : -1;
    return it;
}

/* Returns the next slot in the array, or -1 if there are no more slots. */
int slotRangeArrayNext(slotRangeArrayIter *it) {
    if (it->range_index >= it->slots->num_ranges) return -1;

    if (it->cur_slot < it->slots->ranges[it->range_index].end) {
        it->cur_slot++;
    } else {
        it->range_index++;
        if (it->range_index < it->slots->num_ranges)
            it->cur_slot = it->slots->ranges[it->range_index].start;
        else
            it->cur_slot = -1; /* finished */
    }
    return it->cur_slot;
}

int slotRangeArrayGetCurrentSlot(slotRangeArrayIter *it) {
    return it->cur_slot;
}

void slotRangeArrayIteratorFree(slotRangeArrayIter *it) {
    zfree(it);
}

/* Parse slot range pairs from argv starting at `pos`.
 * `argc` is the argument count, `pos` is the first slot argument index.
 * Returns a slotRangeArray or NULL on error. */
slotRangeArray *parseSlotRangesOrReply(client *c, int argc, int pos) {
    int start, end, count;
    slotRangeArray *slots;

    /* Ensure there is at least one (start,end) slot range pairs. */
    if (argc < 0 || pos < 0 || pos >= argc || (argc - pos) < 2 || ((argc - pos) % 2) != 0) {
        addReplyErrorArity(c);
        return NULL;
    }

    count = (argc - pos) / 2;
    slots = slotRangeArrayCreate(count);
    slots->num_ranges = 0;

    for (int j = pos; j < argc; j += 2) {
        if ((start = getSlotOrReply(c, c->argv[j])) == -1 ||
            (end = getSlotOrReply(c, c->argv[j + 1])) == -1)
        {
            slotRangeArrayFree(slots);
            return NULL;
        }
        slotRangeArraySet(slots, slots->num_ranges, start, end);
        slots->num_ranges++;
    }

    sds err = NULL;
    if (slotRangeArrayNormalizeAndValidate(slots, &err) != C_OK) {
        addReplyErrorSds(c, err);
        slotRangeArrayFree(slots);
        return NULL;
    }
    return slots;
}

/* Return 1 if the keys in the slot can be accessed, 0 otherwise. */
int clusterCanAccessKeysInSlot(int slot) {
    /* If not in cluster mode, all keys are accessible */
    if (server.cluster_enabled == 0) return 1;

    /* If the slot is being imported under old slot migration approach, we should
     * allow to list keys from the slot as previously. */
    if (getImportingSlotSource(slot)) return 1;

    /* If using atomic slot migration, check if the slot belongs to the current
     * node or its master, return 1 if so. */
    clusterNode *myself = getMyClusterNode();
    if (clusterNodeIsSlave(myself)) {
        clusterNode *master = clusterNodeGetMaster(myself);
        if (master && clusterNodeCoversSlot(master, slot))
            return 1;
    } else {
        if (clusterNodeCoversSlot(myself, slot))
            return 1;
    }
    return 0;
}

/* Return the slot ranges that belong to the current node or its master.
 * In non-cluster mode, returns the full slot range (0-16383). */
slotRangeArray *clusterGetLocalSlotRanges(void) {
    if (!server.cluster_enabled) {
        slotRangeArray *slots = slotRangeArrayCreate(1);
        slotRangeArraySet(slots, 0, 0, CLUSTER_SLOTS - 1);
        return slots;
    }

    return clusterGetNodeSlotRanges(getMyClusterNode());
}

/* Returns the slot ranges owned by the given node.
 * If the node is a replica, the master's slot ranges are returned.
 * Returns an empty array if the node has no slots. */
slotRangeArray *clusterGetNodeSlotRanges(clusterNode *node) {
    slotRangeArray *slots = NULL;

    serverAssert(server.cluster_enabled && node != NULL);
    clusterNode *master = clusterNodeGetMaster(node);
    if (master) {
        for (int i = 0; i < CLUSTER_SLOTS; i++) {
            if (clusterNodeCoversSlot(master, i))
                slots = slotRangeArrayAppend(slots, i);
        }
    }
    return slots ? slots : slotRangeArrayCreate(0);
}

/* Partially flush destination DB in a cluster node, based on the slot range.
 *
 * Usage: SFLUSH <start-slot> <end slot> [<start-slot> <end slot>]* [SYNC|ASYNC]
 *
 * Redis will flush the slots that belong to this node and reply with the flushed 
 * slot ranges. If no slot is flushed, an empty array will be returned.
 * 
 * e.g. Node owns slot 100-200, user issues SFLUSH 50 150
 * Redis will flush slot 100-150 and reply with [100,150]
 * 
 * If possible, SFLUSH SYNC will be run as blocking ASYNC as an 
 * optimization.
 */
void sflushCommand(client *c) {
    int flags = EMPTYDB_NO_FLAGS, argc = c->argc;
    int trim_method = ASM_TRIM_METHOD_NONE;

    if (server.cluster_enabled == 0) {
        addReplyError(c,"This instance has cluster support disabled");
        return;
    }

    /* check if last argument is SYNC or ASYNC */
    if (!strcasecmp(c->argv[c->argc-1]->ptr,"sync")) {
        flags = EMPTYDB_NO_FLAGS;
        argc--;
    } else if (!strcasecmp(c->argv[c->argc-1]->ptr,"async")) {
        flags = EMPTYDB_ASYNC;
        argc--;
    } else if (server.lazyfree_lazy_user_flush) {
        flags = EMPTYDB_ASYNC;
    }

    /* parse the slot range */
    if (argc % 2 == 0) {
        addReplyErrorArity(c);
        return;
    }

    /* Parse slot ranges from the command arguments. */
    slotRangeArray *slots = parseSlotRangesOrReply(c, argc, 1);
    if (!slots) return;

    /* If client is AOF or master, we must obey the slot ranges. */
    int must_obey = mustObeyClient(c);

    /* Iterate and find the slot ranges that belong to this node. Save them in
     * a new slotRangeArray. It is allocated on heap since there is a chance
     * that FLUSH SYNC will be running as blocking ASYNC and only later reply
     * with slot ranges */
    slotRangeArray *myslots = NULL;
    for (int i = 0; i < slots->num_ranges; i++) {
        for (int j = slots->ranges[i].start; j <= slots->ranges[i].end; j++) {
            if (must_obey || clusterIsMySlot(j)) {
                myslots = slotRangeArrayAppend(myslots, j);
            }
        }
    }

    /* If no slots belong to this node, return empty array. */
    if (myslots == NULL) {
        addReplyArrayLen(c, 0);
        slotRangeArrayFree(slots);
        return;
    }
    slotRangeArrayFree(slots);
    
    /* takes ownership of myslots */
    asmTrimCtx *trim_ctx = asmTrimCtxCreate(myslots, server.db[0].keys);

    /* If the selected slots are exactly the same as the local slots, we can
     * simply flush the entire DB by flushCommandCommon. */
    slotRangeArray *local_slots = clusterGetLocalSlotRanges();
    int all_slots_covered = slotRangeArrayIsEqual(myslots, local_slots);
    slotRangeArrayFree(local_slots);
    if (all_slots_covered) {
        /* If not flush as blocking async, then reply immediately */
        if (flushCommandCommon(c, FLUSH_TYPE_SLOTS, flags, trim_ctx) == 0) {
            replySlotsFlush(c, trim_ctx->slots);
        }
        asmTrimCtxRelease(trim_ctx);
        return;
    }

    /* Cancel all ASM tasks that overlap with the given slot ranges. */
    clusterAsmCancelBySlotRangeArray(myslots, c->argv[0]->ptr);

    /* In case of SYNC, check if we can optimize and run it in bg as blocking ASYNC */
    int blocking_async = 0;
    if ((!(flags & EMPTYDB_ASYNC)) && (!(c->flags & CLIENT_AVOID_BLOCKING_ASYNC_FLUSH))) {
        flags |= EMPTYDB_ASYNC; /* Run as ASYNC */
        blocking_async = 1;
    }

    /* Trim the slots if running in async mode and not loading from AOF,
     * otherwise delete the keys synchronously. */
    if (flags & EMPTYDB_ASYNC && server.loading == 0) {
        /* Update dirty stats before trimming. */
        server.dirty += getKeyCountInSlotRangeArray(myslots);
        /* Pass client id for active trim to unblock client when trim completes. */
        trim_method = asmTrimSlots(trim_ctx, blocking_async ? c->id : CLIENT_ID_NONE, 0);
    } else {
        clusterDelKeysInSlotRangeArray(myslots, 1);
    }

    /* Without the forceCommandPropagation, when DB was already empty,
     * SFLUSH will not be replicated nor put into the AOF. */
    forceCommandPropagation(c, PROPAGATE_REPL | PROPAGATE_AOF);

    /* Handle waiting for trim job to complete in case of blocking async flush.
     * Block the client and schedule completion callback based on trim method:
     * - BG trim uses BIO lazyfree worker to trim the slots, so schedule a new
     *   BIO lazyfree worker to wait for completion, then unblock client and reply.
     * - Active trim works in cron job of the main thread, it will automatically
     *   unblock client and reply in active trim completion. */
    if (blocking_async && trim_method != ASM_TRIM_METHOD_NONE) {
        blockClientForAsyncFlush(c);
    } else {
        /* Reply with slot ranges that were flushed. SYNC and ASYNC mode will be
         * replied here immediately. */
        replySlotsFlush(c, trim_ctx->slots);
    }

    asmTrimCtxRelease(trim_ctx); /* if bg trim, released later by kvsAsyncFreeDoneCB() */
}

/* The READWRITE command just clears the READONLY command state. */
void readwriteCommand(client *c) {
    if (server.cluster_enabled == 0) {
        addReplyError(c,"This instance has cluster support disabled");
        return;
    }
    c->flags &= ~CLIENT_READONLY;
    addReply(c,shared.ok);
}

/* Resets transient cluster stats that we expose via INFO or other means that we want
 * to reset via CONFIG RESETSTAT. The function is also used in order to
 * initialize these fields in clusterInit() at server startup. */
void resetClusterStats(void) {
    if (!server.cluster_enabled) return;

    clusterSlotStatResetAll();
}

/* This function is called at server startup in order to initialize cluster data
 * structures that are shared between the different cluster implementations. */
void clusterCommonInit(void) {
    resetClusterStats();
    asmInit();
}

/* This function is called after the node startup in order to check if there
 * are any slots that we have keys for, but are not assigned to us. If so,
 * we delete the keys. */
void clusterDeleteKeysInUnownedSlots(void) {
    if (clusterNodeIsSlave(getMyClusterNode())) return;

    /* Check that all the slots we have keys for are assigned to us. Otherwise,
     * delete the keys. */
    for (int i = 0; i < CLUSTER_SLOTS; i++) {
        /* Skip if: no keys in the slot, it's our slot, or we are importing it. */
        if (!countKeysInSlot(i) ||
            clusterIsMySlot(i) ||
            getImportingSlotSource(i))
        {
            continue;
        }

        serverLog(LL_NOTICE, "I have keys for slot %d, but the slot is "
                             "assigned to another node. "
                             "Deleting keys in the slot.", i);
        /* With atomic slot migration, it is safe to drop keys from slots
         * that are not owned. This will not result in data loss under the
         * legacy slot migration approach either, since the importing state
         * has already been persisted in node.conf. */
        clusterDelKeysInSlot(i, 0);
    }
}


/* This function is called after the node startup in order to verify that data
 * loaded from disk is in agreement with the cluster configuration:
 *
 * 1) If we find keys about hash slots we have no responsibility for, the
 *    following happens:
 *    A) If no other node is in charge according to the current cluster
 *       configuration, we add these slots to our node.
 *    B) If according to our config other nodes are already in charge for
 *       this slots, we set the slots as IMPORTING from our point of view
 *       in order to justify we have those slots, and in order to make
 *       redis-cli aware of the issue, so that it can try to fix it.
 * 2) If we find data in a DB different than DB0 we return C_ERR to
 *    signal the caller it should quit the server with an error message
 *    or take other actions.
 *
 * The function always returns C_OK even if it will try to correct
 * the error described in "1". However if data is found in DB different
 * from DB0, C_ERR is returned.
 *
 * The function also uses the logging facility in order to warn the user
 * about desynchronizations between the data we have in memory and the
 * cluster configuration. */
int verifyClusterConfigWithData(void) {
    /* Return ASAP if a module disabled cluster redirections. In that case
     * every master can store keys about every possible hash slot. */
    if (server.cluster_module_flags & CLUSTER_MODULE_FLAG_NO_REDIRECTION)
        return C_OK;

    /* If this node is a slave, don't perform the check at all as we
     * completely depend on the replication stream. */
    if (clusterNodeIsSlave(getMyClusterNode())) return C_OK;

    /* Make sure we only have keys in DB0. */
    for (int i = 1; i < server.dbnum; i++) {
        if (kvstoreSize(server.db[i].keys)) return C_ERR;
    }

    /* Take over slots that we have keys for, but are assigned to no one. */
    clusterClaimUnassignedSlots();
    /* Delete keys in unowned slots */
    clusterDeleteKeysInUnownedSlots();
    return C_OK;
}

/* Notify Redis that the cluster topology changed. (cluster impl --> redis)
 *
 * When CLUSTER_TOPOLOGY_CHANGE_FLAG_SLOT is set, arg points to a
 * slotRangeArray identifying the changed slots. If arg is NULL,
 * all slots are checked against the current topology.
 */
int clusterNotifyTopologyChanged(int flags, void *arg) {
    if (!server.cluster_enabled) return C_OK;
    server.cluster_topology_change_flags |= flags;

    if (flags & CLUSTER_TOPOLOGY_CHANGE_FLAG_SLOT) {
        clusterNode *myself = getMyClusterNode();
        if (!myself) return C_OK; /* The local cluster node is not initialized. */
        clusterNode *master = clusterNodeGetMaster(myself);
        slotRangeArray *changed_slots = arg;
        int num_ranges = changed_slots ? changed_slots->num_ranges : 1;

        for (int i = 0; i < num_ranges; i++) {
            /* Without a changed-slot list, check every slot. */
            int start = changed_slots ? changed_slots->ranges[i].start : 0;
            int end = changed_slots ? changed_slots->ranges[i].end : CLUSTER_SLOTS-1;

            for (int slot = start; slot <= end; slot++) {
                int owned = master && clusterNodeCoversSlot(master, slot);

                /* Per-slot state remains valid while this shard owns the slot. */
                if (owned) continue;

                /* Statistics and subscriptions are stale once this shard no longer
                 * owns the slot. */
                clusterSlotStatReset(slot);
                removeChannelsInSlot(slot);
            }
        }
    }

    /* Cancel ASM tasks that are no longer valid in the updated topology. */
    if (flags & (CLUSTER_TOPOLOGY_CHANGE_FLAG_NODE |
                 CLUSTER_TOPOLOGY_CHANGE_FLAG_ROLE))
    {
        clusterAsmCancelInvalidTasks();
    }

    return C_OK;
}

/* Fire the cluster topology-change notification to modules, if one is pending.
 * Called once per event-loop iteration from clusterCommonBeforeSleep().
 * Topology mutation paths record the relevant reasons, including slot, role,
 * node, and OK/FAIL state changes. This debounces a reshuffle that touches many
 * slots into a single notification, and reports every reason that contributed
 * via the change_flags bitmask in the event data. The check here makes redundant
 * calls a harmless no-op.
 *
 * The cluster first becoming ready is delivered through this same path: the
 * OK/FAIL transition records a STATE reason (slot assignment at startup records
 * a SLOT reason too), so no dedicated "startup" notification is needed. */
static void clusterFireTopologyChangeEventIfNeeded(void) {
    if (!server.cluster_topology_change_flags) return;

    uint64_t module_flags = 0;
    if (server.cluster_topology_change_flags & CLUSTER_TOPOLOGY_CHANGE_FLAG_SLOT)
        module_flags |= REDISMODULE_CLUSTER_TOPOLOGY_CHANGE_FLAG_SLOT;
    if (server.cluster_topology_change_flags & CLUSTER_TOPOLOGY_CHANGE_FLAG_ROLE)
        module_flags |= REDISMODULE_CLUSTER_TOPOLOGY_CHANGE_FLAG_ROLE;
    if (server.cluster_topology_change_flags & CLUSTER_TOPOLOGY_CHANGE_FLAG_STATE)
        module_flags |= REDISMODULE_CLUSTER_TOPOLOGY_CHANGE_FLAG_STATE;
    if (server.cluster_topology_change_flags & CLUSTER_TOPOLOGY_CHANGE_FLAG_NODE)
        module_flags |= REDISMODULE_CLUSTER_TOPOLOGY_CHANGE_FLAG_NODE;

    /* No data payload beyond the change reasons: a subscribing module reads
     * whatever else it needs about the new topology via the cluster info module
     * APIs (CLUSTER SLOTS, RedisModule_GetClusterNodesList/NodeInfo,
     * RedisModule_GetClusterSize, ...). */
    RedisModuleClusterTopologyChangeInfo info = {
        .version = REDISMODULE_CLUSTER_TOPOLOGY_CHANGE_INFO_VERSION,
        .change_flags = module_flags,
    };
    server.cluster_topology_change_flags = 0;
    moduleFireServerEvent(REDISMODULE_EVENT_CLUSTER_TOPOLOGY_CHANGE, 0, &info);
}

/* Handle common cluster work after cluster implementation-specific beforeSleep(). */
void clusterCommonBeforeSleep(void) {
    asmBeforeSleep();
    clusterFireTopologyChangeEventIfNeeded();
}