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

Code before:
    info = sdscatprintf(info, "cluster_stats_messages_sent:%lld\r\n", tot_msg_sent);

    for (int i = 0; i < CLUSTERMSG_TYPE_COUNT; i++) {
        if (server.cluster->stats_bus_messages_received[i] == 0) continue;
        tot_msg_received += server.cluster->stats_bus_messages_received[i];
        info = sdscatprintf(info, "cluster_stats_messages_%s_received:%lld\r\n", clusterGetMessageTypeString(i),
                            server.cluster->stats_bus_messages_received[i]);
    }
    info = sdscatprintf(info, "cluster_stats_messages_received:%lld\r\n", tot_msg_received);

    info = sdscatprintf(info, "total_cluster_links_buffer_limit_exceeded:%llu\r\n",
                        server.cluster->stat_cluster_links_buffer_limit_exceeded);

    return info;
}


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

    pubsubShardUnsubscribeAllChannelsInSlot(slot);
}

/* 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 is_cmd) {
    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;

    kvstoreHashtableIterator *kvs_di = NULL;
    void *next;
    kvs_di = kvstoreGetHashtableIterator(server.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(&server.db[0], key);
        } else {
            dbSyncDelete(&server.db[0], key);
        }
        // if is command, skip del propagate
        if (!is_cmd) propagateDeletion(&server.db[0], key, lazy);
        signalModifiedKey(NULL, &server.db[0], key);
        if (is_cmd) {

Review comment (suggestion):
I like the functionality here but I'm not sure about the name of the `is_cmd` argument. It controls two separate things:

is_cmd false: 

* Propagate the deletions using DEL commands
* Don't send keyspace events

is_cmd true:

* Don't propagate deletions to replicas
* Send keyspace events

To send an implicit FLUSHSLOT command from atomic slot migration when a slot migration is aborted, I guess we want to propagate it as a FLUSHSLOT and not send keyspace events. Replicating FLUSHSLOT can be done using a separate call.

Maybe we should have two separate options instead of `is_cmd`.

Code after:

    for (int i = 0; i < CLUSTERMSG_TYPE_COUNT; i++) {
        if (server.cluster->stats_bus_messages_received[i] == 0) continue;
        tot_msg_received += server.cluster->stats_bus_messages_received[i];
        info = sdscatprintf(info, "cluster_stats_messages_%s_received:%lld\r\n", clusterGetMessageTypeString(i),
                            server.cluster->stats_bus_messages_received[i]);
    }
    info = sdscatprintf(info, "cluster_stats_messages_received:%lld\r\n", tot_msg_received);

    info = sdscatprintf(info, "total_cluster_links_buffer_limit_exceeded:%llu\r\n",
                        server.cluster->stat_cluster_links_buffer_limit_exceeded);

    return info;
}


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

    pubsubShardUnsubscribeAllChannelsInSlot(slot);
}

/* 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);