Unverified Commit 2b88db90 authored by debing.sun's avatar debing.sun Committed by GitHub
Browse files

Fix incorrect lag due to trimming stream via XTRIM command (#13473)

## Describe
When using the `XTRIM` command to trim a stream, it does not update the
maximal tombstone (`max_deleted_entry_id`). This leads to an issue where
the lag calculation incorrectly assumes that there are no tombstones
after the consumer group's last_id, resulting in an inaccurate lag.

The reason XTRIM doesn't need to update the maximal tombstone is that it
always trims from the beginning of the stream. This means that it
consistently changes the position of the first entry, leading to the
following scenarios:

1) First entry trimmed after maximal tombstone:
If the first entry is trimmed to a position after the maximal tombstone,
all tombstones will be before the first entry, so they won't affect the
consumer group's lag.

2) First entry trimmed before maximal tombstone:
If the first entry is trimmed to a position before the maximal
tombstone, the maximal tombstone will not be updated.

## Solution
Therefore, this PR optimizes the lag calculation by ensuring that when
both the consumer group's last_id and the maximal tombstone are behind
the first entry, the consumer group's lag is always equal to the number
of remaining elements in the stream.

Supplement to PR https://github.com/redis/redis/pull/13338
parent b94b714f
......@@ -1441,6 +1441,17 @@ void streamReplyWithCGLag(client *c, stream *s, streamCG *cg) {
/* The lag of a newly-initialized stream is 0. */
lag = 0;
valid = 1;
} else if (!s->length) { /* All entries deleted, now empty. */
lag = 0;
valid = 1;
} else if (streamCompareID(&cg->last_id,&s->first_id) < 0 &&
streamCompareID(&s->max_deleted_entry_id,&s->first_id) < 0)
{
/* When both the consumer group's last_id and the maximum tombstone are behind
* the stream's first entry, the consumer group's lag will always be equal to
* the number of remainin entries in the stream. */
lag = s->length;
valid = 1;
} else if (cg->entries_read != SCG_INVALID_ENTRIES_READ && !streamRangeHasTombstones(s,&cg->last_id,NULL)) {
/* No fragmentation ahead means that the group's logical reads counter
* is valid for performing the lag calculation. */
......
......@@ -1246,15 +1246,25 @@ start_server {
r XREADGROUP GROUP g1 alice STREAMS x > ;# Read one entry
r XADD x 2-0 data c
r XADD x 3-0 data d
r XDEL x 1-0
r XDEL x 2-0
# Now the latest tombstone(2-0) is before the first entry(3-0), but there is still
# a tombstone(2-0) after the last_id of the consume group.
# a tombstone(2-0) after the last_id(1-0) of the consume group.
set reply [r XINFO STREAM x FULL]
set group [lindex [dict get $reply groups] 0]
assert_equal [dict get $group entries-read] 1
assert_equal [dict get $group lag] {}
r XDEL x 1-0
# Although there is a tombstone(2-0) after the consumer group's last_id(1-0), all
# entries before the maximal tombstone have been deleted. This means that both the
# last_id and the largest tombstone are behind the first entry. Therefore, tombstones
# no longer affect the lag, which now reflects the remaining entries in the stream.
set reply [r XINFO STREAM x FULL]
set group [lindex [dict get $reply groups] 0]
assert_equal [dict get $group entries-read] 1
assert_equal [dict get $group lag] 1
# Now there is a tombstone(2-0) after the last_id of the consume group, so after consuming
# entry(3-0), the group's counter will be invalid.
r XREADGROUP GROUP g1 alice STREAMS x >
......@@ -1264,6 +1274,39 @@ start_server {
assert_equal [dict get $group lag] 0
}
test {Consumer group lag with XTRIM} {
r DEL x
r XGROUP CREATE x mygroup $ MKSTREAM
r XADD x 1-0 data a
r XADD x 2-0 data b
r XADD x 3-0 data c
r XADD x 4-0 data d
r XADD x 5-0 data e
r XREADGROUP GROUP mygroup alice COUNT 1 STREAMS x >
set reply [r XINFO STREAM x FULL]
set group [lindex [dict get $reply groups] 0]
assert_equal [dict get $group entries-read] 1
assert_equal [dict get $group lag] 4
# Although XTRIM doesn't update the `max-deleted-entry-id`, it always updates the
# position of the first entry. When trimming causes the first entry to be behind
# the consumer group's last_id, the consumer group's lag will always be equal to
# the number of remainin entries in the stream.
r XTRIM x MAXLEN 1
set reply [r XINFO STREAM x FULL]
set group [lindex [dict get $reply groups] 0]
assert_equal [dict get $reply max-deleted-entry-id] "0-0"
assert_equal [dict get $group entries-read] 1
assert_equal [dict get $group lag] 1
# When all the entries were deleted, the lag is always 0.
r XTRIM x MAXLEN 0
set reply [r XINFO STREAM x FULL]
set group [lindex [dict get $reply groups] 0]
assert_equal [dict get $group lag] 0
}
test {Loading from legacy (Redis <= v6.2.x, rdb_ver < 10) persistence} {
# The payload was DUMPed from a v5 instance after:
# XADD x 1-0 data a
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment