From 4c84f2302b7a3da0a49ea2faed4fc1336ab53caf Mon Sep 17 00:00:00 2001 From: JongsuPark Date: Sun, 19 Jul 2026 07:37:19 +0900 Subject: [PATCH] Stream: clamp group entries_read on XSETID ... ENTRIESADDED XSETID lets you lower a stream's entries_added counter (it only checks that the new value is >= the stream length). It never reconciles that value against the consumer groups' entries_read, so after XTRIM shrinks the stream, entries_added can be set below a group's entries_read. That breaks the invariant entries_read <= entries_added and has two observable effects: 1. XINFO GROUPS reports a negative lag (lag = entries_added - entries_read), which the field must never be. 2. On builds that validate this invariant on load (rdb.c), the resulting RDB/RESTORE payload is rejected with "Stream cgroup entries_read inconsistent with entries_added", so the primary fails to restart, replicas fail to full-sync and backups fail to restore. XGROUP CREATE/SETID already clamp entries_read down to entries_added to keep this invariant; do the same in XSETID when entries_added is lowered. Repro: XADD s * ... (x10); XGROUP CREATE s g 0; XREADGROUP ... COUNT 10 XTRIM s MAXLEN 2; XSETID s ENTRIESADDED 2 XINFO GROUPS s -> lag = -8 (and SAVE + restart fails to load) --- src/t_stream.c | 27 +++++++++++++++++- tests/unit/type/stream-cgroups.tcl | 45 ++++++++++++++++++++++++++++++ 2 files changed, 71 insertions(+), 1 deletion(-) diff --git a/src/t_stream.c b/src/t_stream.c index be1c1d381b4..3887b4a6eb1 100644 --- a/src/t_stream.c +++ b/src/t_stream.c @@ -3810,8 +3810,33 @@ void xsetidCommand(client *c) { } s->last_id = id; - if (entries_added != -1) + if (entries_added != -1) { + uint64_t prev_entries_added = s->entries_added; s->entries_added = entries_added; + /* Lowering entries_added may leave a consumer group's entries_read + * greater than the stream's entries_added. That breaks the lag + * calculation (XINFO GROUPS would report a negative lag) and, on + * builds that validate this invariant, produces an RDB/RESTORE + * payload that fails to load. Clamp each group's counter down, just + * like XGROUP CREATE/SETID does when entries_read is set too high. + * Only needed when entries_added is actually lowered; otherwise the + * entries_read <= entries_added invariant already holds and the loop + * would be a no-op. */ + if (s->entries_added < prev_entries_added && s->cgroups) { + raxIterator ri; + raxStart(&ri, s->cgroups); + raxSeek(&ri, "^", NULL, 0); + while (raxNext(&ri)) { + streamCG *cg = ri.data; + if (cg->entries_read != SCG_INVALID_ENTRIES_READ && + (uint64_t)cg->entries_read > s->entries_added) + { + cg->entries_read = s->entries_added; + } + } + raxStop(&ri); + } + } if (!streamIDEqZero(&max_xdel_id)) s->max_deleted_entry_id = max_xdel_id; addReply(c,shared.ok); diff --git a/tests/unit/type/stream-cgroups.tcl b/tests/unit/type/stream-cgroups.tcl index bcb17f270c1..c1ea3bccba1 100644 --- a/tests/unit/type/stream-cgroups.tcl +++ b/tests/unit/type/stream-cgroups.tcl @@ -4835,3 +4835,48 @@ start_server {tags {"repl external:skip" "stream"}} { } } +start_server {tags {"stream external:skip needs:debug"}} { + # XSETID ... ENTRIESADDED must not leave a consumer group's entries_read + # greater than the stream's entries_added. Otherwise XINFO GROUPS reports a + # negative lag, and the resulting RDB fails to load ("Stream cgroup + # entries_read inconsistent with entries_added"). XGROUP CREATE/SETID + # already clamp entries_read; XSETID must clamp it too. + test "XSETID ENTRIESADDED clamps group entries_read and never yields negative lag" { + r DEL mystream + for {set i 1} {$i <= 10} {incr i} { r XADD mystream * f v$i } + r XGROUP CREATE mystream grp 0 + r XREADGROUP GROUP grp c COUNT 10 STREAMS mystream > + set top [dict get [r XINFO STREAM mystream] last-generated-id] + + # Shrink the stream so entries_added can be lowered below entries_read. + r XTRIM mystream MAXLEN 2 + r XSETID mystream $top ENTRIESADDED 2 + + set ginfo [lindex [r XINFO GROUPS mystream] 0] + assert_equal [dict get $ginfo entries-read] 2 + assert {[dict get $ginfo lag] >= 0} + assert_equal [dict get $ginfo lag] 0 + } + + test "XSETID inconsistency does not produce an unloadable RDB" { + # Reuse the state from the previous test; a reload must succeed and + # keep the counters consistent. + r DEBUG RELOAD + assert_equal [r XLEN mystream] 2 + set ginfo [lindex [r XINFO GROUPS mystream] 0] + assert {[dict get $ginfo lag] >= 0} + assert_equal [dict get $ginfo entries-read] 2 + } + + test "XSETID raising entries_added leaves group entries_read untouched" { + r DEL s2 + for {set i 1} {$i <= 5} {incr i} { r XADD s2 * f v$i } + r XGROUP CREATE s2 g 0 + r XREADGROUP GROUP g c COUNT 3 STREAMS s2 > + set top [dict get [r XINFO STREAM s2] last-generated-id] + r XSETID s2 $top ENTRIESADDED 100 + set ginfo [lindex [r XINFO GROUPS s2] 0] + assert_equal [dict get $ginfo entries-read] 3 + assert_equal [dict get $ginfo lag] 97 + } +}