diff --git a/src/commands.def b/src/commands.def index abf9e7956..8f1bdf50f 100644 --- a/src/commands.def +++ b/src/commands.def @@ -9386,7 +9386,7 @@ struct COMMAND_ARG XGROUP_CREATE_Args[] = { {MAKE_ARG("group",ARG_TYPE_STRING,-1,NULL,NULL,NULL,CMD_ARG_NONE,0,NULL)}, {MAKE_ARG("id-selector",ARG_TYPE_ONEOF,-1,NULL,NULL,NULL,CMD_ARG_NONE,2,NULL),.subargs=XGROUP_CREATE_id_selector_Subargs}, {MAKE_ARG("mkstream",ARG_TYPE_PURE_TOKEN,-1,"MKSTREAM",NULL,NULL,CMD_ARG_OPTIONAL,0,NULL)}, -{MAKE_ARG("entries-read",ARG_TYPE_INTEGER,-1,"ENTRIESREAD",NULL,NULL,CMD_ARG_OPTIONAL,0,NULL)}, +{MAKE_ARG("entriesread",ARG_TYPE_INTEGER,-1,"ENTRIESREAD",NULL,NULL,CMD_ARG_OPTIONAL,0,NULL),.display_text="entries-read"}, }; /********** XGROUP CREATECONSUMER ********************/ diff --git a/src/commands/xgroup-create.json b/src/commands/xgroup-create.json index 6b11a1f00..119d7f300 100644 --- a/src/commands/xgroup-create.json +++ b/src/commands/xgroup-create.json @@ -72,8 +72,9 @@ "optional": true }, { + "name": "entriesread", + "display": "entries-read", "token": "ENTRIESREAD", - "name": "entries-read", "type": "integer", "optional": true } diff --git a/src/t_stream.c b/src/t_stream.c index dda3dab2c..4f9b0eb4b 100644 --- a/src/t_stream.c +++ b/src/t_stream.c @@ -1713,10 +1713,11 @@ size_t streamReplyWithRange(client *c, stream *s, streamID *start, streamID *end group->entries_read = streamEstimateDistanceFromFirstEverEntry(s,&id); } group->last_id = id; - /* Group last ID should be propagated only if NOACK was - * specified, otherwise the last id will be included - * in the propagation of XCLAIM itself. */ - if (noack) propagate_last_id = 1; + /* In the past, we would only set it when NOACK was specified. And in + * #9127, XCLAIM did not propagate entries_read in ACK, which would + * cause entries_read to be inconsistent between master and replicas, + * so here we call streamPropagateGroupID unconditionally. */ + propagate_last_id = 1; } /* Emit a two elements array for each item. The first is diff --git a/tests/unit/multi.tcl b/tests/unit/multi.tcl index 851e02247..4e8e80725 100644 --- a/tests/unit/multi.tcl +++ b/tests/unit/multi.tcl @@ -825,7 +825,9 @@ start_server {tags {"multi"}} { {multi} {xclaim *} {xclaim *} + {xgroup SETID * ENTRIESREAD *} {xclaim *} + {xgroup SETID * ENTRIESREAD *} {exec} } close_replication_stream $repl diff --git a/tests/unit/type/stream-cgroups.tcl b/tests/unit/type/stream-cgroups.tcl index 46e0b05c5..9b457dc67 100644 --- a/tests/unit/type/stream-cgroups.tcl +++ b/tests/unit/type/stream-cgroups.tcl @@ -1277,7 +1277,7 @@ start_server { set replica [srv 0 client] foreach autoclaim {0 1} { - test "Replication tests of XCLAIM with deleted entries (autclaim=$autoclaim)" { + test "Replication tests of XCLAIM with deleted entries (autoclaim=$autoclaim)" { $replica replicaof $master_host $master_port wait_for_condition 50 100 { [s 0 master_link_status] eq {up} @@ -1308,6 +1308,27 @@ start_server { } } } + + test {XREADGROUP ACK would propagate entries-read} { + $master del mystream + $master xadd mystream * a b c d e f + $master xgroup create mystream mygroup $ + $master xreadgroup group mygroup ryan count 1 streams mystream > + $master xadd mystream * a1 b1 a1 b2 + $master xadd mystream * name v1 name v1 + $master xreadgroup group mygroup ryan count 1 streams mystream > + $master xreadgroup group mygroup ryan count 1 streams mystream > + + set reply [$master XINFO STREAM mystream FULL] + set group [lindex [dict get $reply groups] 0] + assert_equal [dict get $group entries-read] 3 + assert_equal [dict get $group lag] 0 + + set reply [$replica XINFO STREAM mystream FULL] + set group [lindex [dict get $reply groups] 0] + assert_equal [dict get $group entries-read] 3 + assert_equal [dict get $group lag] 0 + } } start_server {tags {"stream needs:debug"} overrides {appendonly yes aof-use-rdb-preamble no}} {