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

Code before:
 * if applicable. It also calculates a timeout value (timeout_out) based on an optional
 * timeout argument. If provided, the timeout is added to the current time to obtain an
 * absolute timestamp; if omitted, the default timeout CLUSTER_OPERATION_TIMEOUT is used;
 * if set to 0, it indicates no timeout. The function returns 1 if successful, and 0
 * otherwise, after sending an error message to the client. */
int clusterParseSetSlotCommand(client *c, int *slot_out, clusterNode **node_out, mstime_t *timeout_out) {
    int slot = -1;
    clusterNode *n = NULL;
    mstime_t timeout = commandTimeSnapshot() + CLUSTER_OPERATION_TIMEOUT;
    int optarg_pos = 0;

    /* Allow primaries to replicate "CLUSTER SETSLOT" */
    if (!c->flag.primary && nodeIsReplica(myself)) {
        addReplyError(c, "Please use SETSLOT only with masters.");
        return 0;
    }

    /* If 'myself' is a replica, 'c' must be the primary client. */
    serverAssert(!nodeIsReplica(myself) || c == server.primary);

    if ((slot = getSlotOrReply(c, c->argv[2])) == -1) return 0;

    if (!strcasecmp(c->argv[3]->ptr, "migrating") && c->argc >= 5) {
        /* CLUSTER SETSLOT <SLOT> MIGRATING <NODE> */
        if (nodeIsPrimary(myself) && server.cluster->slots[slot] != myself) {
            addReplyErrorFormat(c, "I'm not the owner of hash slot %u", slot);
            return 0;
        }
        n = clusterLookupNode(c->argv[4]->ptr, sdslen(c->argv[4]->ptr));
        if (n == NULL) {
            addReplyErrorFormat(c, "I don't know about node %s", (char *)c->argv[4]->ptr);
            return 0;
        }
        if (nodeIsReplica(n)) {
            addReplyError(c, "Target node is not a master");
            return 0;
        }
        if (c->argc > 5) optarg_pos = 5;
    } else if (!strcasecmp(c->argv[3]->ptr, "importing") && c->argc >= 5) {
        /* CLUSTER SETSLOT <SLOT> IMPORTING <NODE> */
        if (server.cluster->slots[slot] == myself) {
            addReplyErrorFormat(c, "I'm already the owner of hash slot %u", slot);
            return 0;
        }
        n = clusterLookupNode(c->argv[4]->ptr, sdslen(c->argv[4]->ptr));
        if (n == NULL) {
            addReplyErrorFormat(c, "I don't know about node %s", (char *)c->argv[4]->ptr);
            return 0;
        }
        if (nodeIsReplica(n)) {
            addReplyError(c, "Target node is not a master");

Review comment (suggestion):
Here, the deleted keys are replicated just by multiple DEL commands. For comparison, FLUSHDB is replicated as a single FLUSHDB command, which is much less traffic than many DEL commands. If we want to replicate FLUSHSLOT as a single FLUSHSLOT command, then we should probably check that the replicas supports this though. We already have the replica's version stored in `c->repl_data->replica_version` (REPLCONF VERSION is sent by the replica since 8.0) so we could check that it's >= 8.1 for all connected replicas. It's also fine to skip this. We can do it later as a separate optimization.

Code after:
}

/* Remove all the keys in the specified hash slot.
 * The number of removed items is returned. */
unsigned int delKeysInSlot(unsigned int hashslot, int lazy, bool propagate_del, bool send_del_event) {
    if (!countKeysInSlot(hashslot)) return 0;

    /* We may lose a slot during the pause. We need to track this
     * state so that we don't assert in propagateNow(). */
    server.server_del_keys_in_slot = 1;
    unsigned int j = 0;
    int before_execution_nesting = server.execution_nesting;

    for (int i = 0; i < server.dbnum; i++) {
        kvstoreHashtableIterator *kvs_di = NULL;
        void *next;
        serverDb db = server.db[i];
        kvs_di = kvstoreGetHashtableIterator(db.keys, hashslot, HASHTABLE_ITER_SAFE);
        while (kvstoreHashtableIteratorNext(kvs_di, &next)) {
            robj *valkey = next;
            enterExecutionUnit(1, 0);
            sds sdskey = objectGetKey(valkey);
            robj *key = createStringObject(sdskey, sdslen(sdskey));
            if (lazy) {
                dbAsyncDelete(&db, key);
            } else {
                dbSyncDelete(&db, key);
            }
            // if is command, skip del propagate
            if (propagate_del) propagateDeletion(&db, key, lazy);
            signalModifiedKey(NULL, &db, key);
            if (send_del_event) {
                /* In the `cluster flushslot` scenario, the keys are actually deleted so notify everyone. */
                notifyKeyspaceEvent(NOTIFY_GENERIC, "del", key, db.id);
            } else {
                /* 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, db.id);
            }
            exitExecutionUnit();
            postExecutionUnitOperations();
            decrRefCount(key);
            j++;
            server.dirty++;
        }
        kvstoreReleaseHashtableIterator(kvs_di);
    }
    server.server_del_keys_in_slot = 0;
    serverAssert(server.execution_nesting == before_execution_nesting);
    return j;