Fix unloadable RDB when XSETID lowers entries_added

Project: redis/redis | Language: C | Concepts: consistency-models, error-handling, testing-strategy | Difficulty: mid
Link: https://github.com/redis/redis/pull/15489

The problem

When the XSETID ... ENTRIESADDED command is used to lower a stream's entries_added counter (for example, after trimming the stream), Redis failed to update each consumer group's entries_read counter accordingly. This could result in entries_read > entries_added, which breaks the invariant required for correct lag calculation and RDB loading. As a result, XINFO GROUPS could report negative lag, and an RDB file created in this state would fail to load—rendering backups and restarts impossible.

Why it's non-trivial

A naive fix might simply set entries_added to the requested value, ignoring the relationship with consumer group counters. However, this would leave the system in an inconsistent state, violating invariants relied on by other parts of Redis (e.g., lag calculations, RDB validation). The fix must ensure all related invariants are maintained, and only update group counters when necessary to avoid unnecessary work.

The approach

The main logic change is in xsetidCommand in src/t_stream.c:

-    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);
+        }
+    }

• What it does:
  • If entries_added is changed, and specifically lowered, iterate over all consumer groups for the stream.
  • For each group, if its entries_read is valid and greater than the new entries_added, clamp it down to match.
  • This ensures the invariant entries_read <= entries_added is always maintained, preventing negative lag and RDB load failures.
  • If entries_added is raised, no clamping is performed (since the invariant is not violated).

Test coverage was added in tests/unit/type/stream-cgroups.tcl:

+    test "XSETID ENTRIESADDED clamps group entries_read and keeps the RDB loadable" {
+        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_equal [dict get $ginfo lag] 0
+
+        r DEBUG RELOAD
+        assert_equal [r XLEN mystream] 2
+        set ginfo [lindex [r XINFO GROUPS mystream] 0]
+        assert_equal [dict get $ginfo entries-read] 2
+        assert_equal [dict get $ginfo lag] 0
+    }
+
+    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
+    }

• What it does:
  • Verifies that lowering entries_added clamps entries_read and keeps lag non-negative, even after a reload.
  • Verifies that raising entries_added does not clamp entries_read.

From the review thread

sggeorgiev: Skip the iteration when the new entries_added is >= the old value. The loop is a no-op in this case.

sundb: Should these two tests be combined? We should try to make each test runnable independently.

Takeaways

• Always maintain invariants between related counters or metadata, especially when administrative commands can alter state in ways normal operations cannot.
• When updating state, only perform corrective actions when actually necessary to avoid unnecessary work (e.g., only clamp when lowering, not raising).
• Tests for edge cases should verify both in-memory state and persistence/reload behavior to catch subtle consistency bugs.