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

Code before:

/* Mass-unblock clients because something changed in the instance that makes
 * blocking no longer safe. For example clients blocked in list operations
 * in an instance which turns from primary to replica is unsafe, so this function
 * is called when a primary turns into a replica.
 *
 * The semantics are as follows:
 * - Send a -MOVED to the client in cluster-enabled mode.
 * - Send a -REDIRECT when the client has redirect capability in standalone mode.
 * - Otherwise, send a -UNBLOCKED error to the client while disconnecting it at the same time. */
void disconnectOrRedirectAllBlockedClients(void) {
    listNode *ln;
    listIter li;

    listRewind(server.clients, &li);
    while ((ln = listNext(&li))) {
        client *c = listNodeValue(ln);

        if (c->flag.blocked) {
            /* POSTPONEd clients are an exception, when they'll be unblocked, the
             * command processing will start from scratch, and the command will
             * be either executed or rejected. (unlike LIST blocked clients for
             * which the command is already in progress in a way. */
            if (c->bstate->btype == BLOCKED_POSTPONE) continue;

            if (canRedirectClient(c) && (c->bstate->btype == BLOCKED_LIST || c->bstate->btype == BLOCKED_ZSET ||
                                         c->bstate->btype == BLOCKED_STREAM || c->bstate->btype == BLOCKED_MODULE)) {
                if (c->bstate->btype == BLOCKED_MODULE && !moduleClientIsBlockedOnKeys(c)) continue;
                /* Client has redirect capability and blocked on keys */
                addReplyErrorSds(c, sdscatprintf(sdsempty(), "-REDIRECT %s:%d", server.primary_host, server.primary_port));
                unblockClientOnError(c, NULL);
            } else if (!server.cluster_enabled || clusterRedirectBlockedClientIfNeeded(c)) {
                unblockClientOnError(c, "-UNBLOCKED force unblock from blocking operation, "
                                        "instance state changed (master -> replica?)");
                c->flag.close_after_reply = 1;
            }
        }
    }
}

/* This function should be called by the server every time a single command,
 * a MULTI/EXEC block, or a Lua script, terminated its execution after
 * being called by a client. It handles serving clients blocked in all scenarios
 * where a specific key access requires to block until that key is available.
 *
 * All the keys with at least one client blocked that are signaled as ready
 * are accumulated into the server.ready_keys list. This function will run
 * the list and will serve clients accordingly.
 * Note that the function will iterate again and again (for example as a result of serving BLMOVE
 * we can have new blocking clients to serve because of the PUSH side of BLMOVE.)
 *

Review comment (suggestion):
it's better to distinguish and handle cluster and standalone modes separately:

```c
        if (server.cluster_enabled) {
            if (clusterRedirectBlockedClientIfNeeded(c))
                unblockClient(c);
        } else {
            if (canRedirectClient(c) && (c->bstate->btype == BLOCKED_LIST || c->bstate->btype == BLOCKED_ZSET ||
                                         c->bstate->btype == BLOCKED_STREAM || c->bstate->btype == BLOCKED_MODULE)) {
                if (c->bstate->btype == BLOCKED_MODULE && !moduleClientIsBlockedOnKeys(c)) continue;
                /* Client has redirect capability and blocked on keys */
                addReplyErrorSds(c, sdscatprintf(sdsempty(), "-REDIRECT %s:%d", server.primary_host, server.primary_port));
                unblockClientOnError(c, NULL);
            } else {
                unblockClientOnError(c, "-UNBLOCKED force unblock from blocking operation, "
                                        "instance state changed (master -> replica?)");
                c->flag.close_after_reply = 1;
            }
        }
```

Code after:

/* Mass-unblock clients because something changed in the instance that makes
 * blocking no longer safe. For example clients blocked in list operations
 * in an instance which turns from primary to replica is unsafe, so this function
 * is called when a primary turns into a replica.
 *
 * The semantics are as follows:
 * - If the client is read-only, and blocked by a read command we can handle, we do not unblock it.
 * - Send a -MOVED to the client in cluster-enabled mode.
 * - Send a -REDIRECT when the client has redirect capability in standalone mode.
 * - Otherwise, send a -UNBLOCKED error to the client while disconnecting it at the same time. */
void disconnectOrRedirectAllBlockedClients(void) {
    listNode *ln;
    listIter li;

    listRewind(server.clients, &li);
    while ((ln = listNext(&li))) {
        client *c = listNodeValue(ln);

        if (c->flag.blocked) {
            /* POSTPONEd clients are an exception, when they'll be unblocked, the
             * command processing will start from scratch, and the command will
             * be either executed or rejected. (unlike LIST blocked clients for
             * which the command is already in progress in a way. */
            if (c->bstate->btype == BLOCKED_POSTPONE) continue;

            if (server.cluster_enabled) {
                if (clusterRedirectBlockedClientIfNeeded(c))
                    unblockClient(c, 1);
            } else {
                /* if the client is read-only and blocked by a read command, we do not unblock it */
                if (c->flag.readonly && !(c->lastcmd->flags & CMD_WRITE)) continue;
                if (clientSupportStandAloneRedirect(c) && (c->bstate->btype == BLOCKED_LIST || c->bstate->btype == BLOCKED_ZSET ||
                                                           c->bstate->btype == BLOCKED_STREAM || c->bstate->btype == BLOCKED_MODULE)) {
                    if (c->bstate->btype == BLOCKED_MODULE && !moduleClientIsBlockedOnKeys(c)) continue;
                    /* Client has redirect capability and blocked on keys */
                    addReplyErrorSds(c, sdscatprintf(sdsempty(), "-REDIRECT %s:%d", server.primary_host, server.primary_port));
                    unblockClientOnError(c, NULL);
                } else {
                    unblockClientOnError(c, "-UNBLOCKED force unblock from blocking operation, "
                                            "instance state changed (master -> replica?)");
                    c->flag.close_after_reply = 1;
                }
            }
        }
    }
}

/* This function should be called by the server every time a single command,
 * a MULTI/EXEC block, or a Lua script, terminated its execution after
 * being called by a client. It handles serving clients blocked in all scenarios