Unverified Commit 93fb83b4 authored by debing.sun's avatar debing.sun Committed by GitHub
Browse files

Fix incorrect lag field in XINFO when tombstone is after the last_id of consume group (#13338)

Fix #13337

Ths PR fixes fixed two bugs that caused lag calculation errors.
1. When the latest tombstone is before the first entry, the tombstone
may stil be after the last id of consume group.
2. When a tombstone is after the last id of consume group, the group's
counter will be invalid, we should caculate the entries_read by using
estimates.
parent 94b9072e
......@@ -1405,11 +1405,6 @@ int streamRangeHasTombstones(stream *s, streamID *start, streamID *end) {
return 0;
}
if (streamCompareID(&s->first_id,&s->max_deleted_entry_id) > 0) {
/* The latest tombstone is before the first entry. */
return 0;
}
if (start) {
start_id = *start;
} else {
......@@ -1502,6 +1497,10 @@ long long streamEstimateDistanceFromFirstEverEntry(stream *s, streamID *id) {
return s->entries_added;
}
/* There are fragmentations between the `id` and the stream's last-generated-id. */
if (!streamIDEqZero(id) && streamCompareID(id,&s->max_deleted_entry_id) < 0)
return SCG_INVALID_ENTRIES_READ;
int cmp_last = streamCompareID(id,&s->last_id);
if (cmp_last == 0) {
/* Return the exact counter of the last entry in the stream. */
......@@ -1684,10 +1683,10 @@ size_t streamReplyWithRange(client *c, stream *s, streamID *start, streamID *end
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 && !streamRangeHasTombstones(s,&id,NULL)) {
/* A valid counter and no future tombstones mean we can
* increment the read counter to keep tracking the group's
* progress. */
if (group->entries_read != SCG_INVALID_ENTRIES_READ && !streamRangeHasTombstones(s,&group->last_id,NULL)) {
/* A valid counter and no tombstones between the group's last-delivered-id
* and the stream's last-generated-id mean 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. */
......
......@@ -1239,6 +1239,31 @@ start_server {
assert_equal [dict get $group lag] 2
}
test {Consumer Group Lag with XDELs and tombstone after the last_id of consume group} {
r DEL x
r XGROUP CREATE x g1 $ MKSTREAM
r XADD x 1-0 data a
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.
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] {}
# 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 >
set reply [r XINFO STREAM x FULL]
set group [lindex [dict get $reply groups] 0]
assert_equal [dict get $group entries-read] 3
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