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

Code before:
     * thread_shared_qb, we still add this check for code robustness. */
    int use_thread_shared_qb = (c->querybuf == thread_shared_qb) ? 1 : 0;
    if (!is_primary && // primary client's querybuf can grow greedy.
        (big_arg || sdsalloc(c->querybuf) < PROTO_IOBUF_LEN)) {
        /* When reading a BIG_ARG we won't be reading more than that one arg
         * into the query buffer, so we don't need to pre-allocate more than we
         * need, so using the non-greedy growing. For an initial allocation of
         * the query buffer, we also don't wanna use the greedy growth, in order
         * to avoid collision with the RESIZE_THRESHOLD mechanism. */
        c->querybuf = sdsMakeRoomForNonGreedy(c->querybuf, readlen);
        /* We later set the peak to the used portion of the buffer, but here we over
         * allocated because we know what we need, make sure it'll not be shrunk before used. */
        if (c->querybuf_peak < qblen + readlen) c->querybuf_peak = qblen + readlen;
    } else {
        c->querybuf = sdsMakeRoomFor(c->querybuf, readlen);

        /* Read as much as possible from the socket to save read(2) system calls. */
        readlen = sdsavail(c->querybuf);
    }
    if (use_thread_shared_qb) serverAssert(c->querybuf == thread_shared_qb);

    c->nread = connRead(c->conn, c->querybuf + qblen, readlen);
    if (c->nread <= 0) {
        return;
    }
    c->is_qb_full_read = (size_t)c->nread == readlen ? 1 : 0;

    sdsIncrLen(c->querybuf, c->nread);
    qblen = sdslen(c->querybuf);
    if (c->querybuf_peak < qblen) c->querybuf_peak = qblen;
    if (!is_primary) {
        /* The commands cached in the MULTI/EXEC queue have not been executed yet,
         * so they are also considered a part of the query buffer in a broader sense.
         *
         * For unauthenticated clients, the query buffer cannot exceed 1MB at most. */
        size_t qb_memory = sdslen(c->querybuf) + (c->mstate ? c->mstate->argv_len_sums : 0);
        if (qb_memory > server.client_max_querybuf_len ||
            (qb_memory > 1024 * 1024 && (c->read_flags & READ_FLAGS_AUTH_REQUIRED))) {
            c->read_flags |= READ_FLAGS_QB_LIMIT_REACHED;
        }
    }
}

/**
 * This function is designed to prioritize replication flow.
 * Determines whether the replica should continue reading from the primary.
 * It dynamically adjusts the read rate based on buffer utilization
 *
 * @return          1 if another read should be attempted, 0 otherwise.
 */
int shouldRepeatRead(client *c, int iteration) {

Review comment (suggestion):
```suggestion
    c->is_qb_full_read = (size_t)c->nread == readlen;
```
This is a bool.

Code after:
    /* Don't process input from the primary while there is a busy script
     * condition on the replica. We want just to accumulate the replication
     * stream (instead of replying -BUSY like we do with other clients) and
     * later resume the processing. */
    if (isInsideYieldingLongCommand() && c->flag.primary) return 0;

    /* CLIENT_CLOSE_AFTER_REPLY closes the connection once the reply is
     * written to the client. Make sure to not let the reply grow after
     * this flag has been set (i.e. don't process more commands).
     *
     * The same applies for clients we want to terminate ASAP. */
    if (c->flag.close_after_reply || c->flag.close_asap) return 0;

    return 1;
}

int processInputBuffer(client *c) {
    /* Parse the query buffer. */
    while (c->querybuf && c->qb_pos < sdslen(c->querybuf)) {
        if (!canParseCommand(c)) {
            break;
        }

        c->read_flags = c->flag.primary ? READ_FLAGS_PRIMARY : 0;
        c->read_flags |= authRequired(c) ? READ_FLAGS_AUTH_REQUIRED : 0;

        parseCommand(c);

        if (handleParseResults(c) != PARSE_OK) {
            break;
        }

        if (c->argc == 0) {
            /* No command to process - continue parsing the query buf. */
            continue;
        }

        if (c->querybuf == thread_shared_qb) {
            /* Before processing the command, reset the shared query buffer to its default state.
             * This avoids unintentionally modifying the shared qb during processCommand as we may use
             * the shared qb for other clients during processEventsWhileBlocked */
            resetSharedQueryBuf(c);
        }

        /* We are finally ready to execute the command. */
        if (processCommandAndResetClient(c) == C_ERR) {
            /* If the client is no longer valid, we avoid exiting this
             * loop and trimming the client buffer later. So we return
             * ASAP in that case. */
            return C_ERR;
        }