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

Code before:
                            stream *s,
                            streamID *start,
                            streamID *end,
                            size_t count,
                            int rev,
                            streamCG *group,
                            streamConsumer *consumer,
                            int flags,
                            streamPropInfo *spi) {
    void *arraylen_ptr = NULL;
    size_t arraylen = 0;
    streamIterator si;
    int64_t numfields;
    streamID id;
    int propagate_last_id = 0;
    int noack = flags & STREAM_RWR_NOACK;

    /* If the client is asking for some history, we serve it using a
     * different function, so that we return entries *solely* from its
     * own PEL. This ensures each consumer will always and only see
     * the history of messages delivered to it and not yet confirmed
     * as delivered. */
    if (group && (flags & STREAM_RWR_HISTORY)) {
        return streamReplyWithRangeFromConsumerPEL(c, s, start, end, count, consumer);
    }

    if (!(flags & STREAM_RWR_RAWENTRIES)) arraylen_ptr = addReplyDeferredLen(c);
    streamIteratorStart(&si, s, start, end, rev);
    while (streamIteratorGetID(&si, &id, &numfields)) {
        /* Update the group last_id if needed. */
        if (group && streamCompareID(&id, &group->last_id) > 0) {
            if (group->entries_read != SCG_INVALID_ENTRIES_READ && streamCompareID(&group->last_id, &s->first_id) >= 0 && !streamRangeHasTombstones(s, &group->last_id, NULL)) {
                /* A valid counter and no tombstones in the group's last-delivered-id and the stream's last-generated-id,
                 * we can increment the read counter to keep tracking the group's progress. */
                group->entries_read++;
            } else if (s->entries_added) {
                /* The group's counter may be invalid, so we try to obtain it. */
                group->entries_read = streamEstimateDistanceFromFirstEverEntry(s, &id);
            }
            group->last_id = id;
            /* In the past, we would only set it when NOACK was specified. And in
             * #9127, XCLAIM did not propagate entries_read in ACK, which would
             * cause entries_read to be inconsistent between primary and replicas,
             * so here we call streamPropagateGroupID unconditionally. */
            propagate_last_id = 1;
        }

        /* Emit a two elements array for each item. The first is
         * the ID, the second is an array of field-value pairs. */
        addReplyArrayLen(c, 2);
        addReplyStreamID(c, &id);

Review comment (bug):
streamCompareID(&group->last_id, &s->first_id) >= 0

This judgment will result in most of them failing to perform, ultimately causing performance degradation.

Causes most requests to recalculate the distance.


Code after:
                            streamConsumer *consumer,
                            int flags,
                            streamPropInfo *spi) {
    void *arraylen_ptr = NULL;
    size_t arraylen = 0;
    streamIterator si;
    int64_t numfields;
    streamID id;
    int propagate_last_id = 0;
    int noack = flags & STREAM_RWR_NOACK;

    /* If the client is asking for some history, we serve it using a
     * different function, so that we return entries *solely* from its
     * own PEL. This ensures each consumer will always and only see
     * the history of messages delivered to it and not yet confirmed
     * as delivered. */
    if (group && (flags & STREAM_RWR_HISTORY)) {
        return streamReplyWithRangeFromConsumerPEL(c, s, start, end, count, consumer);
    }

    if (!(flags & STREAM_RWR_RAWENTRIES)) arraylen_ptr = addReplyDeferredLen(c);
    streamIteratorStart(&si, s, start, end, rev);
    while (streamIteratorGetID(&si, &id, &numfields)) {
        /* Update the group last_id if needed. */
        if (group && streamCompareID(&id, &group->last_id) > 0) {
            if (group->entries_read != SCG_INVALID_ENTRIES_READ &&
                streamCompareID(&group->last_id, &s->first_id) >= 0 &&
                !streamRangeHasTombstones(s, &group->last_id, NULL)) {
                /* A valid counter and no tombstones in the group's last-delivered-id and the stream's last-generated-id,
                 * we can increment the read counter to keep tracking the group's progress. */
                group->entries_read++;
            } else if (s->entries_added) {
                /* The group's counter may be invalid, so we try to obtain it. */
                group->entries_read = streamEstimateDistanceFromFirstEverEntry(s, &id);
            }
            group->last_id = id;
            /* In the past, we would only set it when NOACK was specified. And in
             * #9127, XCLAIM did not propagate entries_read in ACK, which would
             * cause entries_read to be inconsistent between primary and replicas,
             * so here we call streamPropagateGroupID unconditionally. */
            propagate_last_id = 1;
        }

        /* Emit a two elements array for each item. The first is
         * the ID, the second is an array of field-value pairs. */
        addReplyArrayLen(c, 2);
        addReplyStreamID(c, &id);

        addReplyArrayLen(c, numfields * 2);

        /* Emit the field-value pairs. */