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

Code before:
        resetClient(c);
    }

    /* We count blocked client stats on regular clients and not on module clients */
    if (!c->flag.module) server.blocked_clients--;
    server.blocked_clients_by_type[c->bstate->btype]--;
    /* Clear the flags, and put the client in the unblocked list so that
     * we'll process new commands in its query buffer ASAP. */
    c->flag.blocked = 0;
    c->bstate->btype = BLOCKED_NONE;
    c->bstate->unblock_on_nokey = 0;
    removeClientFromTimeoutTable(c);
    if (queue_for_reprocessing) queueClientForReprocessing(c);
}

/* This function gets called when a blocked client timed out in order to
 * send it a reply of some kind. After this function is called,
 * unblockClient() will be called with the same client as argument. */
void replyToBlockedClientTimedOut(client *c) {
    if (c->bstate->btype == BLOCKED_LIST || c->bstate->btype == BLOCKED_ZSET || c->bstate->btype == BLOCKED_STREAM) {
        addReplyNullArray(c);
        updateStatsOnUnblock(c, 0, 0, 0);
    } else if (c->bstate->btype == BLOCKED_WAIT) {
        if (c->cmd->proc == waitCommand) {
            addReplyLongLong(c, replicationCountAcksByOffset(c->bstate->reploffset));
        } else if (c->cmd->proc == waitaofCommand) {
            addReplyArrayLen(c, 2);
            addReplyLongLong(c, server.fsynced_reploff >= c->bstate->reploffset);
            addReplyLongLong(c, replicationCountAOFAcksByOffset(c->bstate->reploffset));
        } else if (c->cmd->proc == clusterCommand) {
            addReplyErrorObject(c, shared.noreplicaserr);
        } else {
            serverPanic("Unknown wait command %s in replyToBlockedClientTimedOut().", c->cmd->declared_name);
        }
    } else if (c->bstate->btype == BLOCKED_MODULE) {
        moduleBlockedClientTimedOut(c, 0);
    } else {
        serverPanic("Unknown btype in replyToBlockedClientTimedOut().");
    }
}

/* If one or more clients are blocked on the SHUTDOWN command, this function
 * sends them an error reply and unblocks them. */
void replyToClientsBlockedOnShutdown(void) {
    if (server.blocked_clients_by_type[BLOCKED_SHUTDOWN] == 0) return;
    listNode *ln;
    listIter li;
    listRewind(server.clients, &li);
    while ((ln = listNext(&li))) {
        client *c = listNodeValue(ln);
        if (c->flag.blocked && c->bstate->btype == BLOCKED_SHUTDOWN) {

Review comment (suggestion):
Not that important, but maybe we should check the client has a defined timeout? The downside is that it would make this a much more "breaking change" but the logic would be much more correct IMO since if a client application blocked a client without timeout it might also not expect to be timedout  

Code after:
        resetClient(c);
    }

    /* We count blocked client stats on regular clients and not on module clients */
    if (!c->flag.module) server.blocked_clients--;
    server.blocked_clients_by_type[c->bstate->btype]--;
    /* Clear the flags, and put the client in the unblocked list so that
     * we'll process new commands in its query buffer ASAP. */
    c->flag.blocked = 0;
    c->bstate->btype = BLOCKED_NONE;
    c->bstate->unblock_on_nokey = 0;
    removeClientFromTimeoutTable(c);
    if (queue_for_reprocessing) queueClientForReprocessing(c);
}

/* Check if the specified client can be safely timed out using
 * unblockClientOnTimeout(). */
int blockedClientMayTimeout(client *c) {
    if (c->bstate->btype == BLOCKED_MODULE) {
        return moduleBlockedClientMayTimeout(c);
    }

    if (c->bstate->btype == BLOCKED_LIST ||
        c->bstate->btype == BLOCKED_ZSET ||
        c->bstate->btype == BLOCKED_STREAM ||
        c->bstate->btype == BLOCKED_WAIT) {
        return 1;
    }
    return 0;
}

/* This function gets called when a blocked client timed out in order to
 * send it a reply of some kind. After this function is called,
 * unblockClient() will be called with the same client as argument. */
void replyToBlockedClientTimedOut(client *c) {
    if (c->bstate->btype == BLOCKED_LIST || c->bstate->btype == BLOCKED_ZSET || c->bstate->btype == BLOCKED_STREAM) {
        addReplyNullArray(c);
        updateStatsOnUnblock(c, 0, 0, 0);
    } else if (c->bstate->btype == BLOCKED_WAIT) {
        if (c->cmd->proc == waitCommand) {
            addReplyLongLong(c, replicationCountAcksByOffset(c->bstate->reploffset));
        } else if (c->cmd->proc == waitaofCommand) {
            addReplyArrayLen(c, 2);
            addReplyLongLong(c, server.fsynced_reploff >= c->bstate->reploffset);
            addReplyLongLong(c, replicationCountAOFAcksByOffset(c->bstate->reploffset));
        } else if (c->cmd->proc == clusterCommand) {
            addReplyErrorObject(c, shared.noreplicaserr);
        } else {
            serverPanic("Unknown wait command %s in replyToBlockedClientTimedOut().", c->cmd->declared_name);
        }
    } else if (c->bstate->btype == BLOCKED_MODULE) {