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

Code before:
    if (c->flag.replica && (is_may_replicate_command || is_write_command || is_read_command)) {
        rejectCommandFormat(c, "Replica can't interact with the keyspace");
        return C_OK;
    }

    /* If the server is paused, block the client until
     * the pause has ended. Replicas are never paused. */
    if (!c->flag.replica && ((isPausedActions(PAUSE_ACTION_CLIENT_ALL)) ||
                             ((isPausedActions(PAUSE_ACTION_CLIENT_WRITE)) && is_may_replicate_command))) {
        blockPostponeClient(c);
        return C_OK;
    }

    /* Exec the command */
    if (c->flag.multi && c->cmd->proc == clientReplyCommand) {
        addReplyError(c, "CLIENT REPLY not allowed during MULTI/EXEC transaction.");
    } else if (c->flag.multi && c->cmd->proc != execCommand && c->cmd->proc != discardCommand &&
               c->cmd->proc != multiCommand && c->cmd->proc != watchCommand && c->cmd->proc != quitCommand &&
               c->cmd->proc != resetCommand) {
        queueMultiCommand(c, cmd_flags);
        addReply(c, shared.queued);
    } else {
        int flags = CMD_CALL_FULL;
        /* we should re-tag the client as reprocessing in case it was reprocessing the same command */
        if (client_reprocessing_command)
            c->flag.reprocessing_command = 1;
        call(c, flags);
        if (listLength(server.ready_keys) && !isInsideYieldingLongCommand()) handleClientsBlockedOnKeys();
    }
    return C_OK;
}

/* ====================== Error lookup and execution ===================== */

void incrementErrorCount(const char *fullerr, size_t namelen) {
    void *result;
    if (!raxFind(server.errors, (unsigned char *)fullerr, namelen, &result)) {
        struct serverError *error = zmalloc(sizeof(*error));
        error->count = 1;
        raxInsert(server.errors, (unsigned char *)fullerr, namelen, error, NULL);
    } else {
        struct serverError *error = result;
        error->count++;
    }
}

/*================================== Shutdown =============================== */

/* Close listening sockets. Also unlink the unix domain socket if
 * unlink_unix_socket is non-zero. */
void closeListeningSockets(int unlink_unix_socket) {

Review comment (suggestion):
we added this line because this logic? We can also remove CMD_CALL_REPROCESSING flag since it is uesless now. Or we can keep it if we keep this logic and use CMD_CALL_REPROCESSING flag and then use CMD_CALL_REPROCESSING flag in call to flag the client flag.
```
    /* we should clear this flag, since it is possible the processing will bail out due to some error
     * and we do not want to keep this flag on for the next commands */
    c->flag.reprocessing_command = 0;
```

so the all changes, that will something like this:
```diff
diff --git a/src/blocked.c b/src/blocked.c
index 42d1fa641..606f3fbd6 100644
--- a/src/blocked.c
+++ b/src/blocked.c
@@ -680,6 +680,7 @@ static void unblockClientOnKey(client *c, robj *key) {
      * we need to re process the command again */
     if (c->flag.pending_command) {
         c->flag.pending_command = 0;
+        c->flag.reprocessing_command = 1;
         /* We want the command processing and the unblock handler (see RM_Call 'K' option)
          * to run atomically, this is why we must enter the execution unit here before
          * running the command, and exit the execution unit after calling the unblock handler (if exists).
@@ -699,6 +700,7 @@ static void unblockClientOnKey(client *c, robj *key) {
         exitExecutionUnit();
         afterCommand(c);
         server.current_client = old_client;
+        c->flag.reprocessing_command = 0;
     }
 }
 
diff --git a/src/server.c b/src/server.c
index f7bc6e766..9622483b2 100644
--- a/src/server.c
+++ b/src/server.c
@@ -4031,7 +4031,10 @@ int processCommand(client *c) {
 
     /* in case we are starting to ProcessCommand and we already have a command we assume
      * this is a reprocessing of this command, so we do not want to perform some of the actions again. */
-    int client_reprocessing_command = c->cmd ? 1 : 0;
+    int client_reprocessing_command = c->flag.reprocessing_command ? 1 : 0;
+    /* we should clear this flag, since it is possible the processing will bail out due to some error
+     * and we do not want to keep this flag on for the next commands */
+    c->flag.reprocessing_command = 0;
 
     /* only run command filter if not reprocessing command */
     if (!client_reprocessing_command) {

```

Code after:
    if (!c->flag.replica && ((isPausedActions(PAUSE_ACTION_CLIENT_ALL)) ||
                             ((isPausedActions(PAUSE_ACTION_CLIENT_WRITE)) && is_may_replicate_command))) {
        blockPostponeClient(c);
        return C_OK;
    }

    /* Exec the command */
    if (c->flag.multi && c->cmd->proc == clientReplyCommand) {
        addReplyError(c, "CLIENT REPLY not allowed during MULTI/EXEC transaction.");
    } else if (c->flag.multi && c->cmd->proc != execCommand && c->cmd->proc != discardCommand &&
               c->cmd->proc != multiCommand && c->cmd->proc != watchCommand && c->cmd->proc != quitCommand &&
               c->cmd->proc != resetCommand) {
        queueMultiCommand(c, cmd_flags);
        addReply(c, shared.queued);
    } else {
        int flags = CMD_CALL_FULL;
        call(c, flags);
        if (listLength(server.ready_keys) && !isInsideYieldingLongCommand()) handleClientsBlockedOnKeys();
    }
    return C_OK;
}

/* ====================== Error lookup and execution ===================== */

void incrementErrorCount(const char *fullerr, size_t namelen) {
    void *result;
    if (!raxFind(server.errors, (unsigned char *)fullerr, namelen, &result)) {
        struct serverError *error = zmalloc(sizeof(*error));
        error->count = 1;
        raxInsert(server.errors, (unsigned char *)fullerr, namelen, error, NULL);
    } else {
        struct serverError *error = result;
        error->count++;
    }
}

/*================================== Shutdown =============================== */

/* Close listening sockets. Also unlink the unix domain socket if
 * unlink_unix_socket is non-zero. */
void closeListeningSockets(int unlink_unix_socket) {
    int j;

    for (int i = 0; i < CONN_TYPE_MAX; i++) {
        connListener *listener = &server.listeners[i];
        if (listener->ct == NULL) continue;

        for (j = 0; j < listener->count; j++) close(listener->fd[j]);
    }

    if (server.cluster_enabled)