Skip to content

Commit c6cc724

Browse files
artikellartikell
committed
Fix incorrect lag and entries-read value with tombstone
Signed-off-by: artikell <[email protected]>
1 parent 28e5dcc commit c6cc724

File tree

2 files changed

+133
-68
lines changed

2 files changed

+133
-68
lines changed

‎src/t_stream.c‎

Lines changed: 41 additions & 59 deletions
Original file line numberDiff line numberDiff line change
@@ -64,7 +64,7 @@ size_t streamReplyWithRangeFromConsumerPEL(client *c,
6464
streamConsumer *consumer);
6565
int streamParseStrictIDOrReply(client *c, robj *o, streamID *id, uint64_t missing_seq, int *seq_given);
6666
int streamParseIDOrReply(client *c, robj *o, streamID *id, uint64_t missing_seq);
67-
67+
static long long streamEstimateDistance(stream *s, streamCG *cg, streamID *next_id);
6868
/* -----------------------------------------------------------------------
6969
* Low level stream encoding: a radix tree of listpacks.
7070
* ----------------------------------------------------------------------- */
@@ -1389,11 +1389,6 @@ int streamIDEqZero(streamID *id) {
13891389
int streamRangeHasTombstones(stream *s, streamID *start, streamID *end) {
13901390
streamID start_id, end_id;
13911391

1392-
if (!s->length || streamIDEqZero(&s->max_deleted_entry_id)) {
1393-
/* The stream is empty or has no tombstones. */
1394-
return 0;
1395-
}
1396-
13971392
if (start) {
13981393
start_id = *start;
13991394
} else {
@@ -1418,39 +1413,56 @@ int streamRangeHasTombstones(stream *s, streamID *start, streamID *end) {
14181413
return 0;
14191414
}
14201415

1421-
/* Replies with a consumer group's current lag, that is the number of messages
1422-
* in the stream that are yet to be delivered. In case that the lag isn't
1423-
* available due to fragmentation, the reply to the client is a null. */
1424-
void streamReplyWithCGLag(client *c, stream *s, streamCG *cg) {
1416+
/* Replies with a consumer group's current lag, which is the number of messages in the stream
1417+
* that are yet to be delivered. Additionally, it includes an entries-read field that indicates
1418+
* the number of messages currently read. In case that the lag or entries-read isn't available
1419+
* due to fragmentation, the reply to the client is null. */
1420+
void streamReplyWithCGLagAndEntriesRead(client *c, stream *s, streamCG *cg) {
14251421
int valid = 0;
14261422
long long lag = 0;
14271423

1428-
if (!s->entries_added) {
1429-
/* The lag of a newly-initialized stream is 0. */
1430-
lag = 0;
1424+
/* Attempt to retrieve the group's last ID logical read counter. */
1425+
long long entries_read = streamEstimateDistance(s, cg, &cg->last_id);
1426+
if (entries_read != SCG_INVALID_ENTRIES_READ) {
1427+
/* A valid counter was obtained. */
1428+
lag = (long long)s->entries_added - entries_read;
14311429
valid = 1;
1432-
} else if (cg->entries_read != SCG_INVALID_ENTRIES_READ && !streamRangeHasTombstones(s, &cg->last_id, NULL)) {
1433-
/* No fragmentation ahead means that the group's logical reads counter
1434-
* is valid for performing the lag calculation. */
1435-
lag = (long long)s->entries_added - cg->entries_read;
1436-
valid = 1;
1437-
} else {
1438-
/* Attempt to retrieve the group's last ID logical read counter. */
1439-
long long entries_read = streamEstimateDistanceFromFirstEverEntry(s, &cg->last_id);
1440-
if (entries_read != SCG_INVALID_ENTRIES_READ) {
1441-
/* A valid counter was obtained. */
1442-
lag = (long long)s->entries_added - entries_read;
1443-
valid = 1;
1444-
}
14451430
}
14461431

14471432
if (valid) {
1433+
/* Read counter of the last delivered ID */
1434+
addReplyBulkCString(c, "entries-read");
1435+
addReplyLongLong(c, entries_read);
1436+
/* Group lag */
1437+
addReplyBulkCString(c, "lag");
14481438
addReplyLongLong(c, lag);
14491439
} else {
1440+
addReplyBulkCString(c, "entries-read");
1441+
addReplyNull(c);
1442+
addReplyBulkCString(c, "lag");
14501443
addReplyNull(c);
14511444
}
14521445
}
14531446

1447+
/* The function returns the logical read counter corresponding to next_id
1448+
* based on the information of the group.
1449+
*/
1450+
static long long streamEstimateDistance(stream *s, streamCG *cg, streamID *next_id) {
1451+
/* If the values of next_id and last_id are the same,
1452+
* it is considered that only the current value needs to be returned,
1453+
* otherwise it is considered to be the calculated value.
1454+
* This is used to align with the streamEstimateDistanceFromFirstEverEntry method.
1455+
*/
1456+
long long step = streamCompareID(&cg->last_id, next_id) == 0 ? 0 : 1;
1457+
if (cg->entries_read != SCG_INVALID_ENTRIES_READ && !streamRangeHasTombstones(s, &cg->last_id, NULL)) {
1458+
/* A valid counter and no future tombstones mean we can
1459+
* increment the read counter to keep tracking the group's
1460+
* progress. */
1461+
return cg->entries_read + step;
1462+
}
1463+
return streamEstimateDistanceFromFirstEverEntry(s, next_id);
1464+
}
1465+
14541466
/* This function returns a value that is the ID's logical read counter, or its
14551467
* distance (the number of entries) from the first entry ever to have been added
14561468
* to the stream.
@@ -1485,11 +1497,6 @@ long long streamEstimateDistanceFromFirstEverEntry(stream *s, streamID *id) {
14851497
return s->entries_added;
14861498
}
14871499

1488-
if (!streamIDEqZero(id) && streamCompareID(id, &s->max_deleted_entry_id) < 0) {
1489-
/* The ID is before the last tombstone, so the counter is unknown. */
1490-
return SCG_INVALID_ENTRIES_READ;
1491-
}
1492-
14931500
int cmp_last = streamCompareID(id, &s->last_id);
14941501
if (cmp_last == 0) {
14951502
/* Return the exact counter of the last entry in the stream. */
@@ -1678,16 +1685,7 @@ size_t streamReplyWithRange(client *c,
16781685
while (streamIteratorGetID(&si, &id, &numfields)) {
16791686
/* Update the group last_id if needed. */
16801687
if (group && streamCompareID(&id, &group->last_id) > 0) {
1681-
if (group->entries_read != SCG_INVALID_ENTRIES_READ &&
1682-
streamCompareID(&group->last_id, &s->first_id) >= 0 &&
1683-
!streamRangeHasTombstones(s, &group->last_id, NULL)) {
1684-
/* A valid counter and no tombstones in the group's last-delivered-id and the stream's last-generated-id,
1685-
* we can increment the read counter to keep tracking the group's progress. */
1686-
group->entries_read++;
1687-
} else if (s->entries_added) {
1688-
/* The group's counter may be invalid, so we try to obtain it. */
1689-
group->entries_read = streamEstimateDistanceFromFirstEverEntry(s, &id);
1690-
}
1688+
group->entries_read = streamEstimateDistance(s, group, &id);
16911689
group->last_id = id;
16921690
/* In the past, we would only set it when NOACK was specified. And in
16931691
* #9127, XCLAIM did not propagate entries_read in ACK, which would
@@ -3691,16 +3689,7 @@ void xinfoReplyWithStreamInfo(client *c, stream *s) {
36913689
addReplyStreamID(c, &cg->last_id);
36923690

36933691
/* Read counter of the last delivered ID */
3694-
addReplyBulkCString(c, "entries-read");
3695-
if (cg->entries_read != SCG_INVALID_ENTRIES_READ) {
3696-
addReplyLongLong(c, cg->entries_read);
3697-
} else {
3698-
addReplyNull(c);
3699-
}
3700-
3701-
/* Group lag */
3702-
addReplyBulkCString(c, "lag");
3703-
streamReplyWithCGLag(c, s, cg);
3692+
streamReplyWithCGLagAndEntriesRead(c, s, cg);
37043693

37053694
/* Group PEL count */
37063695
addReplyBulkCString(c, "pel-count");
@@ -3887,14 +3876,7 @@ void xinfoCommand(client *c) {
38873876
addReplyLongLong(c, raxSize(cg->pel));
38883877
addReplyBulkCString(c, "last-delivered-id");
38893878
addReplyStreamID(c, &cg->last_id);
3890-
addReplyBulkCString(c, "entries-read");
3891-
if (cg->entries_read != SCG_INVALID_ENTRIES_READ) {
3892-
addReplyLongLong(c, cg->entries_read);
3893-
} else {
3894-
addReplyNull(c);
3895-
}
3896-
addReplyBulkCString(c, "lag");
3897-
streamReplyWithCGLag(c, s, cg);
3879+
streamReplyWithCGLagAndEntriesRead(c, s, cg);
38983880
}
38993881
raxStop(&ri);
39003882
} else if (!strcasecmp(opt, "STREAM")) {

‎tests/unit/type/stream-cgroups.tcl‎

Lines changed: 92 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -1065,7 +1065,7 @@ start_server {
10651065
set group [lindex [dict get $reply groups] 0]
10661066
assert_equal [dict get $reply max-deleted-entry-id] "0-0"
10671067
assert_equal [dict get $reply entries-added] 0
1068-
assert_equal [dict get $group entries-read] {}
1068+
assert_equal [dict get $group entries-read] 0
10691069
assert_equal [dict get $group lag] 0
10701070

10711071
r XADD x 1-0 data a
@@ -1075,7 +1075,7 @@ start_server {
10751075
set group [lindex [dict get $reply groups] 0]
10761076
assert_equal [dict get $reply max-deleted-entry-id] "1-0"
10771077
assert_equal [dict get $reply entries-added] 1
1078-
assert_equal [dict get $group entries-read] {}
1078+
assert_equal [dict get $group entries-read] 1
10791079
assert_equal [dict get $group lag] 0
10801080
}
10811081

@@ -1090,7 +1090,7 @@ start_server {
10901090

10911091
set reply [r XINFO STREAM x FULL]
10921092
set group [lindex [dict get $reply groups] 0]
1093-
assert_equal [dict get $group entries-read] {}
1093+
assert_equal [dict get $group entries-read] 0
10941094
assert_equal [dict get $group lag] 5
10951095

10961096
r XREADGROUP GROUP g1 c11 COUNT 1 STREAMS x >
@@ -1121,7 +1121,7 @@ start_server {
11211121
# so the lag can't be calculated
11221122
set reply [r XINFO STREAM x FULL]
11231123
set group [lindex [dict get $reply groups] 0]
1124-
assert_equal [dict get $group entries-read] 8
1124+
assert_equal [dict get $group entries-read] {}
11251125
assert_equal [dict get $group lag] {}
11261126

11271127
r XREADGROUP GROUP g1 c12 COUNT 1 STREAMS x >
@@ -1183,7 +1183,7 @@ start_server {
11831183
assert_equal [dict get $group entries-read] 5
11841184
assert_equal [dict get $group lag] 1
11851185
set group [lindex [dict get $reply groups] 1]
1186-
assert_equal [dict get $group entries-read] {}
1186+
assert_equal [dict get $group entries-read] 3
11871187
assert_equal [dict get $group lag] 3
11881188

11891189
r XTRIM x MINID = 5-0
@@ -1192,7 +1192,7 @@ start_server {
11921192
assert_equal [dict get $group entries-read] 5
11931193
assert_equal [dict get $group lag] 1
11941194
set group [lindex [dict get $reply groups] 1]
1195-
assert_equal [dict get $group entries-read] {}
1195+
assert_equal [dict get $group entries-read] 4
11961196
assert_equal [dict get $group lag] 2
11971197
}
11981198

@@ -1249,7 +1249,7 @@ start_server {
12491249
assert_equal [dict get $group entries-read] 5
12501250
assert_equal [dict get $group lag] 2
12511251
set group [lindex [dict get $reply groups] 1]
1252-
assert_equal [dict get $group entries-read] {}
1252+
assert_equal [dict get $group entries-read] 6
12531253
assert_equal [dict get $group lag] 1
12541254

12551255
r XREADGROUP GROUP g1 c11 STREAMS x >
@@ -1259,7 +1259,7 @@ start_server {
12591259
assert_equal [dict get $group lag] 0
12601260
}
12611261

1262-
test {Consumer group lag with XADD trimming} {
1262+
test {Consumer group lag with XADD trimming} {
12631263
r DEL x
12641264
r XADD x 1-0 data a
12651265
r XADD x 2-0 data b
@@ -1305,14 +1305,17 @@ start_server {
13051305
set group [lindex [dict get $reply groups] 0]
13061306
assert_equal [dict get $group entries-read] 5
13071307
assert_equal [dict get $group lag] 2
1308+
set group [lindex [dict get $reply groups] 1]
1309+
assert_equal [dict get $group entries-read] {}
1310+
assert_equal [dict get $group lag] {}
13081311

13091312
r XADD x MINID = 7-0 8-0 data h
13101313
set reply [r XINFO STREAM x FULL]
13111314
set group [lindex [dict get $reply groups] 0]
13121315
assert_equal [dict get $group entries-read] 5
13131316
assert_equal [dict get $group lag] 3
13141317
set group [lindex [dict get $reply groups] 1]
1315-
assert_equal [dict get $group entries-read] {}
1318+
assert_equal [dict get $group entries-read] 6
13161319
assert_equal [dict get $group lag] 2
13171320

13181321
r XREADGROUP GROUP g1 c11 STREAMS x >
@@ -1322,6 +1325,86 @@ start_server {
13221325
assert_equal [dict get $group lag] 0
13231326
}
13241327

1328+
test {Consumer group lag with with tombstone} {
1329+
r DEL x
1330+
r XGROUP CREATE x processing $ MKSTREAM
1331+
r XADD x 0-1 name Mercury
1332+
r XREADGROUP GROUP processing alice STREAMS x >
1333+
r XADD x 0-2 name Venus
1334+
r XADD x 0-3 name Earth
1335+
r XADD x 0-4 name Jupiter
1336+
r XADD x 0-5 name Jupiter
1337+
1338+
set reply [r XINFO STREAM x FULL]
1339+
set group [lindex [dict get $reply groups] 0]
1340+
assert_equal [dict get $group entries-read] 1
1341+
assert_equal [dict get $group lag] 4
1342+
1343+
r XDEL x 0-1
1344+
set reply [r XINFO STREAM x FULL]
1345+
set group [lindex [dict get $reply groups] 0]
1346+
assert_equal [dict get $group entries-read] 1
1347+
assert_equal [dict get $group lag] 4
1348+
1349+
r XDEL x 0-2
1350+
set reply [r XINFO STREAM x FULL]
1351+
set group [lindex [dict get $reply groups] 0]
1352+
assert_equal [dict get $group entries-read] 2
1353+
assert_equal [dict get $group lag] 3
1354+
1355+
r XDEL x 0-3
1356+
set reply [r XINFO STREAM x FULL]
1357+
set group [lindex [dict get $reply groups] 0]
1358+
assert_equal [dict get $group entries-read] 3
1359+
assert_equal [dict get $group lag] 2
1360+
1361+
r XDEL x 0-4
1362+
set reply [r XINFO STREAM x FULL]
1363+
set group [lindex [dict get $reply groups] 0]
1364+
assert_equal [dict get $group entries-read] 4
1365+
assert_equal [dict get $group lag] 1
1366+
1367+
r XDEL x 0-5
1368+
set reply [r XINFO STREAM x FULL]
1369+
set group [lindex [dict get $reply groups] 0]
1370+
assert_equal [dict get $group entries-read] 5
1371+
assert_equal [dict get $group lag] 0
1372+
}
1373+
1374+
test {Consumer group check lag and entries-read consistency} {
1375+
r DEL x
1376+
r XGROUP CREATE x processing $ MKSTREAM
1377+
r XGROUP CREATE x processing1 $ MKSTREAM
1378+
r XADD x 0-1 name Mercury
1379+
r XADD x 0-2 name Venus
1380+
r XADD x 0-3 name Earth
1381+
r XADD x 0-4 name Jupiter
1382+
1383+
r XREADGROUP GROUP processing alice COUNT 2 STREAMS x >
1384+
r XDEL x 0-3
1385+
1386+
set reply [r XINFO STREAM x FULL]
1387+
set group [lindex [dict get $reply groups] 0]
1388+
assert_equal [dict get $group entries-read] {}
1389+
assert_equal [dict get $group lag] {}
1390+
1391+
r DEL x
1392+
r XGROUP CREATE x processing $ MKSTREAM
1393+
r XGROUP CREATE x processing1 $ MKSTREAM
1394+
r XADD x 0-1 name Mercury
1395+
r XADD x 0-2 name Venus
1396+
r XADD x 0-3 name Earth
1397+
r XADD x 0-4 name Jupiter
1398+
1399+
r XDEL x 0-3
1400+
r XREADGROUP GROUP processing alice COUNT 2 STREAMS x >
1401+
1402+
set reply [r XINFO STREAM x FULL]
1403+
set group [lindex [dict get $reply groups] 0]
1404+
assert_equal [dict get $group entries-read] {}
1405+
assert_equal [dict get $group lag] {}
1406+
}
1407+
13251408
test {Loading from legacy (Redis <= v6.2.x, rdb_ver < 10) persistence} {
13261409
# The payload was DUMPed from a v5 instance after:
13271410
# XADD x 1-0 data a

0 commit comments

Comments
 (0)