diff --git a/src/commands.def b/src/commands.def index bd19c4cd6..9bb55c4c2 100644 --- a/src/commands.def +++ b/src/commands.def @@ -9649,6 +9649,46 @@ struct COMMAND_ARG XACK_Args[] = { {MAKE_ARG("id",ARG_TYPE_STRING,-1,NULL,NULL,NULL,CMD_ARG_MULTIPLE,0,NULL)}, }; +/********** XACKDEL ********************/ + +#ifndef SKIP_CMD_HISTORY_TABLE +/* XACKDEL history */ +#define XACKDEL_History NULL +#endif + +#ifndef SKIP_CMD_TIPS_TABLE +/* XACKDEL tips */ +#define XACKDEL_Tips NULL +#endif + +#ifndef SKIP_CMD_KEY_SPECS_TABLE +/* XACKDEL key specs */ +keySpec XACKDEL_Keyspecs[1] = { +{NULL,CMD_KEY_RW|CMD_KEY_UPDATE|CMD_KEY_DELETE,KSPEC_BS_INDEX,.bs.index={1},KSPEC_FK_RANGE,.fk.range={0,1,0}} +}; +#endif + +/* XACKDEL condition argument table */ +struct COMMAND_ARG XACKDEL_condition_Subargs[] = { +{MAKE_ARG("keepref",ARG_TYPE_PURE_TOKEN,-1,"KEEPREF",NULL,NULL,CMD_ARG_NONE,0,NULL)}, +{MAKE_ARG("delref",ARG_TYPE_PURE_TOKEN,-1,"DELREF",NULL,NULL,CMD_ARG_NONE,0,NULL)}, +{MAKE_ARG("acked",ARG_TYPE_PURE_TOKEN,-1,"ACKED",NULL,NULL,CMD_ARG_NONE,0,NULL)}, +}; + +/* XACKDEL ids argument table */ +struct COMMAND_ARG XACKDEL_ids_Subargs[] = { +{MAKE_ARG("numids",ARG_TYPE_INTEGER,-1,NULL,NULL,NULL,CMD_ARG_NONE,0,NULL)}, +{MAKE_ARG("id",ARG_TYPE_STRING,-1,NULL,NULL,NULL,CMD_ARG_MULTIPLE,0,NULL)}, +}; + +/* XACKDEL argument table */ +struct COMMAND_ARG XACKDEL_Args[] = { +{MAKE_ARG("key",ARG_TYPE_KEY,0,NULL,NULL,NULL,CMD_ARG_NONE,0,NULL)}, +{MAKE_ARG("group",ARG_TYPE_STRING,-1,NULL,NULL,NULL,CMD_ARG_NONE,0,NULL)}, +{MAKE_ARG("condition",ARG_TYPE_ONEOF,-1,NULL,NULL,NULL,CMD_ARG_OPTIONAL,3,NULL),.subargs=XACKDEL_condition_Subargs}, +{MAKE_ARG("ids",ARG_TYPE_BLOCK,-1,"IDS",NULL,NULL,CMD_ARG_NONE,2,NULL),.subargs=XACKDEL_ids_Subargs}, +}; + /********** XADD ********************/ #ifndef SKIP_CMD_HISTORY_TABLE @@ -9656,6 +9696,7 @@ struct COMMAND_ARG XACK_Args[] = { commandHistory XADD_History[] = { {"6.2.0","Added the `NOMKSTREAM` option, `MINID` trimming strategy and the `LIMIT` option."}, {"7.0.0","Added support for the `-*` explicit ID form."}, +{"8.2.0","Added the `KEEPREF`, `DELREF` and `ACKED` options."}, }; #endif @@ -9673,6 +9714,13 @@ keySpec XADD_Keyspecs[1] = { }; #endif +/* XADD condition argument table */ +struct COMMAND_ARG XADD_condition_Subargs[] = { +{MAKE_ARG("keepref",ARG_TYPE_PURE_TOKEN,-1,"KEEPREF",NULL,NULL,CMD_ARG_NONE,0,NULL)}, +{MAKE_ARG("delref",ARG_TYPE_PURE_TOKEN,-1,"DELREF",NULL,NULL,CMD_ARG_NONE,0,NULL)}, +{MAKE_ARG("acked",ARG_TYPE_PURE_TOKEN,-1,"ACKED",NULL,NULL,CMD_ARG_NONE,0,NULL)}, +}; + /* XADD trim strategy argument table */ struct COMMAND_ARG XADD_trim_strategy_Subargs[] = { {MAKE_ARG("maxlen",ARG_TYPE_PURE_TOKEN,-1,"MAXLEN",NULL,NULL,CMD_ARG_NONE,0,NULL)}, @@ -9709,6 +9757,7 @@ struct COMMAND_ARG XADD_data_Subargs[] = { struct COMMAND_ARG XADD_Args[] = { {MAKE_ARG("key",ARG_TYPE_KEY,0,NULL,NULL,NULL,CMD_ARG_NONE,0,NULL)}, {MAKE_ARG("nomkstream",ARG_TYPE_PURE_TOKEN,-1,"NOMKSTREAM",NULL,"6.2.0",CMD_ARG_OPTIONAL,0,NULL)}, +{MAKE_ARG("condition",ARG_TYPE_ONEOF,-1,NULL,NULL,NULL,CMD_ARG_OPTIONAL,3,NULL),.subargs=XADD_condition_Subargs}, {MAKE_ARG("trim",ARG_TYPE_BLOCK,-1,NULL,NULL,NULL,CMD_ARG_OPTIONAL,4,NULL),.subargs=XADD_trim_Subargs}, {MAKE_ARG("id-selector",ARG_TYPE_ONEOF,-1,NULL,NULL,NULL,CMD_ARG_NONE,2,NULL),.subargs=XADD_id_selector_Subargs}, {MAKE_ARG("data",ARG_TYPE_BLOCK,-1,NULL,NULL,NULL,CMD_ARG_MULTIPLE,2,NULL),.subargs=XADD_data_Subargs}, @@ -9809,6 +9858,45 @@ struct COMMAND_ARG XDEL_Args[] = { {MAKE_ARG("id",ARG_TYPE_STRING,-1,NULL,NULL,NULL,CMD_ARG_MULTIPLE,0,NULL)}, }; +/********** XDELEX ********************/ + +#ifndef SKIP_CMD_HISTORY_TABLE +/* XDELEX history */ +#define XDELEX_History NULL +#endif + +#ifndef SKIP_CMD_TIPS_TABLE +/* XDELEX tips */ +#define XDELEX_Tips NULL +#endif + +#ifndef SKIP_CMD_KEY_SPECS_TABLE +/* XDELEX key specs */ +keySpec XDELEX_Keyspecs[1] = { +{NULL,CMD_KEY_RW|CMD_KEY_DELETE,KSPEC_BS_INDEX,.bs.index={1},KSPEC_FK_RANGE,.fk.range={0,1,0}} +}; +#endif + +/* XDELEX condition argument table */ +struct COMMAND_ARG XDELEX_condition_Subargs[] = { +{MAKE_ARG("keepref",ARG_TYPE_PURE_TOKEN,-1,"KEEPREF",NULL,NULL,CMD_ARG_NONE,0,NULL)}, +{MAKE_ARG("delref",ARG_TYPE_PURE_TOKEN,-1,"DELREF",NULL,NULL,CMD_ARG_NONE,0,NULL)}, +{MAKE_ARG("acked",ARG_TYPE_PURE_TOKEN,-1,"ACKED",NULL,NULL,CMD_ARG_NONE,0,NULL)}, +}; + +/* XDELEX ids argument table */ +struct COMMAND_ARG XDELEX_ids_Subargs[] = { +{MAKE_ARG("numids",ARG_TYPE_INTEGER,-1,NULL,NULL,NULL,CMD_ARG_NONE,0,NULL)}, +{MAKE_ARG("id",ARG_TYPE_STRING,-1,NULL,NULL,NULL,CMD_ARG_MULTIPLE,0,NULL)}, +}; + +/* XDELEX argument table */ +struct COMMAND_ARG XDELEX_Args[] = { +{MAKE_ARG("key",ARG_TYPE_KEY,0,NULL,NULL,NULL,CMD_ARG_NONE,0,NULL)}, +{MAKE_ARG("condition",ARG_TYPE_ONEOF,-1,NULL,NULL,NULL,CMD_ARG_OPTIONAL,3,NULL),.subargs=XDELEX_condition_Subargs}, +{MAKE_ARG("ids",ARG_TYPE_BLOCK,-1,"IDS",NULL,NULL,CMD_ARG_NONE,2,NULL),.subargs=XDELEX_ids_Subargs}, +}; + /********** XGROUP CREATE ********************/ #ifndef SKIP_CMD_HISTORY_TABLE @@ -10363,6 +10451,7 @@ struct COMMAND_ARG XSETID_Args[] = { /* XTRIM history */ commandHistory XTRIM_History[] = { {"6.2.0","Added the `MINID` trimming strategy and the `LIMIT` option."}, +{"8.2.0","Added the `KEEPREF`, `DELREF` and `ACKED` options."}, }; #endif @@ -10392,18 +10481,26 @@ struct COMMAND_ARG XTRIM_trim_operator_Subargs[] = { {MAKE_ARG("approximately",ARG_TYPE_PURE_TOKEN,-1,"~",NULL,NULL,CMD_ARG_NONE,0,NULL)}, }; +/* XTRIM trim condition argument table */ +struct COMMAND_ARG XTRIM_trim_condition_Subargs[] = { +{MAKE_ARG("keepref",ARG_TYPE_PURE_TOKEN,-1,"KEEPREF",NULL,NULL,CMD_ARG_NONE,0,NULL)}, +{MAKE_ARG("delref",ARG_TYPE_PURE_TOKEN,-1,"DELREF",NULL,NULL,CMD_ARG_NONE,0,NULL)}, +{MAKE_ARG("acked",ARG_TYPE_PURE_TOKEN,-1,"ACKED",NULL,NULL,CMD_ARG_NONE,0,NULL)}, +}; + /* XTRIM trim argument table */ struct COMMAND_ARG XTRIM_trim_Subargs[] = { {MAKE_ARG("strategy",ARG_TYPE_ONEOF,-1,NULL,NULL,NULL,CMD_ARG_NONE,2,NULL),.subargs=XTRIM_trim_strategy_Subargs}, {MAKE_ARG("operator",ARG_TYPE_ONEOF,-1,NULL,NULL,NULL,CMD_ARG_OPTIONAL,2,NULL),.subargs=XTRIM_trim_operator_Subargs}, {MAKE_ARG("threshold",ARG_TYPE_STRING,-1,NULL,NULL,NULL,CMD_ARG_NONE,0,NULL)}, {MAKE_ARG("count",ARG_TYPE_INTEGER,-1,"LIMIT",NULL,"6.2.0",CMD_ARG_OPTIONAL,0,NULL)}, +{MAKE_ARG("condition",ARG_TYPE_ONEOF,-1,NULL,NULL,NULL,CMD_ARG_OPTIONAL,3,NULL),.subargs=XTRIM_trim_condition_Subargs}, }; /* XTRIM argument table */ struct COMMAND_ARG XTRIM_Args[] = { {MAKE_ARG("key",ARG_TYPE_KEY,0,NULL,NULL,NULL,CMD_ARG_NONE,0,NULL)}, -{MAKE_ARG("trim",ARG_TYPE_BLOCK,-1,NULL,NULL,NULL,CMD_ARG_NONE,4,NULL),.subargs=XTRIM_trim_Subargs}, +{MAKE_ARG("trim",ARG_TYPE_BLOCK,-1,NULL,NULL,NULL,CMD_ARG_NONE,5,NULL),.subargs=XTRIM_trim_Subargs}, }; /********** APPEND ********************/ @@ -11328,10 +11425,12 @@ struct COMMAND_STRUCT redisCommandTable[] = { {MAKE_CMD("zunionstore","Stores the union of multiple sorted sets in a key.","O(N)+O(M log(M)) with N being the sum of the sizes of the input sorted sets, and M being the number of elements in the resulting sorted set.","2.0.0",CMD_DOC_NONE,NULL,NULL,"sorted_set",COMMAND_GROUP_SORTED_SET,ZUNIONSTORE_History,0,ZUNIONSTORE_Tips,0,zunionstoreCommand,-4,CMD_WRITE|CMD_DENYOOM,ACL_CATEGORY_SORTEDSET,ZUNIONSTORE_Keyspecs,2,zunionInterDiffStoreGetKeys,5),.args=ZUNIONSTORE_Args}, /* stream */ {MAKE_CMD("xack","Returns the number of messages that were successfully acknowledged by the consumer group member of a stream.","O(1) for each message ID processed.","5.0.0",CMD_DOC_NONE,NULL,NULL,"stream",COMMAND_GROUP_STREAM,XACK_History,0,XACK_Tips,0,xackCommand,-4,CMD_WRITE|CMD_FAST,ACL_CATEGORY_STREAM,XACK_Keyspecs,1,NULL,3),.args=XACK_Args}, -{MAKE_CMD("xadd","Appends a new message to a stream. Creates the key if it doesn't exist.","O(1) when adding a new entry, O(N) when trimming where N being the number of entries evicted.","5.0.0",CMD_DOC_NONE,NULL,NULL,"stream",COMMAND_GROUP_STREAM,XADD_History,2,XADD_Tips,1,xaddCommand,-5,CMD_WRITE|CMD_DENYOOM|CMD_FAST,ACL_CATEGORY_STREAM,XADD_Keyspecs,1,NULL,5),.args=XADD_Args}, +{MAKE_CMD("xackdel","Acknowledges and deletes one or multiple messages for a stream consumer group.","O(1) for each message ID processed.","8.2.0",CMD_DOC_NONE,NULL,NULL,"stream",COMMAND_GROUP_STREAM,XACKDEL_History,0,XACKDEL_Tips,0,xackdelCommand,-6,CMD_WRITE|CMD_FAST,ACL_CATEGORY_STREAM,XACKDEL_Keyspecs,1,NULL,4),.args=XACKDEL_Args}, +{MAKE_CMD("xadd","Appends a new message to a stream. Creates the key if it doesn't exist.","O(1) when adding a new entry, O(N) when trimming where N being the number of entries evicted.","5.0.0",CMD_DOC_NONE,NULL,NULL,"stream",COMMAND_GROUP_STREAM,XADD_History,3,XADD_Tips,1,xaddCommand,-5,CMD_WRITE|CMD_DENYOOM|CMD_FAST,ACL_CATEGORY_STREAM,XADD_Keyspecs,1,NULL,6),.args=XADD_Args}, {MAKE_CMD("xautoclaim","Changes, or acquires, ownership of messages in a consumer group, as if the messages were delivered to as consumer group member.","O(1) if COUNT is small.","6.2.0",CMD_DOC_NONE,NULL,NULL,"stream",COMMAND_GROUP_STREAM,XAUTOCLAIM_History,1,XAUTOCLAIM_Tips,1,xautoclaimCommand,-6,CMD_WRITE|CMD_FAST,ACL_CATEGORY_STREAM,XAUTOCLAIM_Keyspecs,1,NULL,7),.args=XAUTOCLAIM_Args}, {MAKE_CMD("xclaim","Changes, or acquires, ownership of a message in a consumer group, as if the message was delivered a consumer group member.","O(log N) with N being the number of messages in the PEL of the consumer group.","5.0.0",CMD_DOC_NONE,NULL,NULL,"stream",COMMAND_GROUP_STREAM,XCLAIM_History,0,XCLAIM_Tips,1,xclaimCommand,-6,CMD_WRITE|CMD_FAST,ACL_CATEGORY_STREAM,XCLAIM_Keyspecs,1,NULL,11),.args=XCLAIM_Args}, {MAKE_CMD("xdel","Returns the number of messages after removing them from a stream.","O(1) for each single item to delete in the stream, regardless of the stream size.","5.0.0",CMD_DOC_NONE,NULL,NULL,"stream",COMMAND_GROUP_STREAM,XDEL_History,0,XDEL_Tips,0,xdelCommand,-3,CMD_WRITE|CMD_FAST,ACL_CATEGORY_STREAM,XDEL_Keyspecs,1,NULL,2),.args=XDEL_Args}, +{MAKE_CMD("xdelex","Deletes one or multiple entries from the stream.","O(1) for each single item to delete in the stream, regardless of the stream size.","8.2.0",CMD_DOC_NONE,NULL,NULL,"stream",COMMAND_GROUP_STREAM,XDELEX_History,0,XDELEX_Tips,0,xdelexCommand,-5,CMD_WRITE|CMD_FAST,ACL_CATEGORY_STREAM,XDELEX_Keyspecs,1,NULL,3),.args=XDELEX_Args}, {MAKE_CMD("xgroup","A container for consumer groups commands.","Depends on subcommand.","5.0.0",CMD_DOC_NONE,NULL,NULL,"stream",COMMAND_GROUP_STREAM,XGROUP_History,0,XGROUP_Tips,0,NULL,-2,0,0,XGROUP_Keyspecs,0,NULL,0),.subcommands=XGROUP_Subcommands}, {MAKE_CMD("xinfo","A container for stream introspection commands.","Depends on subcommand.","5.0.0",CMD_DOC_NONE,NULL,NULL,"stream",COMMAND_GROUP_STREAM,XINFO_History,0,XINFO_Tips,0,NULL,-2,0,0,XINFO_Keyspecs,0,NULL,0),.subcommands=XINFO_Subcommands}, {MAKE_CMD("xlen","Return the number of messages in a stream.","O(1)","5.0.0",CMD_DOC_NONE,NULL,NULL,"stream",COMMAND_GROUP_STREAM,XLEN_History,0,XLEN_Tips,0,xlenCommand,2,CMD_READONLY|CMD_FAST,ACL_CATEGORY_STREAM,XLEN_Keyspecs,1,NULL,1),.args=XLEN_Args}, @@ -11341,7 +11440,7 @@ struct COMMAND_STRUCT redisCommandTable[] = { {MAKE_CMD("xreadgroup","Returns new or historical messages from a stream for a consumer in a group. Blocks until a message is available otherwise.","For each stream mentioned: O(M) with M being the number of elements returned. If M is constant (e.g. always asking for the first 10 elements with COUNT), you can consider it O(1). On the other side when XREADGROUP blocks, XADD will pay the O(N) time in order to serve the N clients blocked on the stream getting new data.","5.0.0",CMD_DOC_NONE,NULL,NULL,"stream",COMMAND_GROUP_STREAM,XREADGROUP_History,0,XREADGROUP_Tips,0,xreadCommand,-7,CMD_BLOCKING|CMD_WRITE,ACL_CATEGORY_STREAM,XREADGROUP_Keyspecs,1,xreadGetKeys,5),.args=XREADGROUP_Args}, {MAKE_CMD("xrevrange","Returns the messages from a stream within a range of IDs in reverse order.","O(N) with N being the number of elements returned. If N is constant (e.g. always asking for the first 10 elements with COUNT), you can consider it O(1).","5.0.0",CMD_DOC_NONE,NULL,NULL,"stream",COMMAND_GROUP_STREAM,XREVRANGE_History,1,XREVRANGE_Tips,0,xrevrangeCommand,-4,CMD_READONLY,ACL_CATEGORY_STREAM,XREVRANGE_Keyspecs,1,NULL,4),.args=XREVRANGE_Args}, {MAKE_CMD("xsetid","An internal command for replicating stream values.","O(1)","5.0.0",CMD_DOC_NONE,NULL,NULL,"stream",COMMAND_GROUP_STREAM,XSETID_History,1,XSETID_Tips,0,xsetidCommand,-3,CMD_WRITE|CMD_DENYOOM|CMD_FAST,ACL_CATEGORY_STREAM,XSETID_Keyspecs,1,NULL,4),.args=XSETID_Args}, -{MAKE_CMD("xtrim","Deletes messages from the beginning of a stream.","O(N), with N being the number of evicted entries. Constant times are very small however, since entries are organized in macro nodes containing multiple entries that can be released with a single deallocation.","5.0.0",CMD_DOC_NONE,NULL,NULL,"stream",COMMAND_GROUP_STREAM,XTRIM_History,1,XTRIM_Tips,1,xtrimCommand,-4,CMD_WRITE,ACL_CATEGORY_STREAM,XTRIM_Keyspecs,1,NULL,2),.args=XTRIM_Args}, +{MAKE_CMD("xtrim","Deletes messages from the beginning of a stream.","O(N), with N being the number of evicted entries. Constant times are very small however, since entries are organized in macro nodes containing multiple entries that can be released with a single deallocation.","5.0.0",CMD_DOC_NONE,NULL,NULL,"stream",COMMAND_GROUP_STREAM,XTRIM_History,2,XTRIM_Tips,1,xtrimCommand,-4,CMD_WRITE,ACL_CATEGORY_STREAM,XTRIM_Keyspecs,1,NULL,2),.args=XTRIM_Args}, /* string */ {MAKE_CMD("append","Appends a string to the value of a key. Creates the key if it doesn't exist.","O(1). The amortized time complexity is O(1) assuming the appended value is small and the already present value is of any size, since the dynamic string library used by Redis will double the free space available on every reallocation.","2.0.0",CMD_DOC_NONE,NULL,NULL,"string",COMMAND_GROUP_STRING,APPEND_History,0,APPEND_Tips,0,appendCommand,3,CMD_WRITE|CMD_DENYOOM|CMD_FAST,ACL_CATEGORY_STRING,APPEND_Keyspecs,1,NULL,2),.args=APPEND_Args}, {MAKE_CMD("decr","Decrements the integer value of a key by one. Uses 0 as initial value if the key doesn't exist.","O(1)","1.0.0",CMD_DOC_NONE,NULL,NULL,"string",COMMAND_GROUP_STRING,DECR_History,0,DECR_Tips,0,decrCommand,2,CMD_WRITE|CMD_DENYOOM|CMD_FAST,ACL_CATEGORY_STRING,DECR_Keyspecs,1,NULL,1),.args=DECR_Args}, diff --git a/src/commands/xackdel.json b/src/commands/xackdel.json new file mode 100644 index 000000000..f33d128ba --- /dev/null +++ b/src/commands/xackdel.json @@ -0,0 +1,109 @@ +{ + "XACKDEL": { + "summary": "Acknowledges and deletes one or multiple messages for a stream consumer group.", + "complexity": "O(1) for each message ID processed.", + "group": "stream", + "since": "8.2.0", + "arity": -6, + "function": "xackdelCommand", + "command_flags": [ + "WRITE", + "FAST" + ], + "acl_categories": [ + "STREAM" + ], + "key_specs": [ + { + "flags": [ + "RW", + "UPDATE", + "DELETE" + ], + "begin_search": { + "index": { + "pos": 1 + } + }, + "find_keys": { + "range": { + "lastkey": 0, + "step": 1, + "limit": 0 + } + } + } + ], + "arguments": [ + { + "name": "key", + "type": "key", + "key_spec_index": 0 + }, + { + "name": "group", + "type": "string" + }, + { + "name": "condition", + "type": "oneof", + "optional": true, + "arguments": [ + { + "name": "keepref", + "type": "pure-token", + "token": "KEEPREF" + }, + { + "name": "delref", + "type": "pure-token", + "token": "DELREF" + }, + { + "name": "acked", + "type": "pure-token", + "token": "ACKED" + } + ] + }, + { + "name": "ids", + "token": "IDS", + "type": "block", + "arguments": [ + { + "name": "numids", + "type": "integer" + }, + { + "name": "id", + "type": "string", + "multiple": true + } + ] + } + ], + "reply_schema": { + "description": "Array of results. Returns an array with -1 for each requested ID if the key does not exist.", + "type": "array", + "minItems": 0, + "maxItems": 4294967295, + "items": { + "oneOf": [ + { + "description": "The id does not exist in the provided stream key.", + "const": -1 + }, + { + "description": "Entry was acknowledged and deleted from the stream.", + "const": 1 + }, + { + "description": "Entry was acknowledged but not deleted, there are still dangling references.", + "const": 2 + } + ] + } + } + } +} diff --git a/src/commands/xadd.json b/src/commands/xadd.json index 21880524e..92d37d438 100644 --- a/src/commands/xadd.json +++ b/src/commands/xadd.json @@ -14,6 +14,10 @@ [ "7.0.0", "Added support for the `-*` explicit ID form." + ], + [ + "8.2.0", + "Added the `KEEPREF`, `DELREF` and `ACKED` options." ] ], "command_flags": [ @@ -61,6 +65,28 @@ "optional": true, "since": "6.2.0" }, + { + "name": "condition", + "type": "oneof", + "optional": true, + "arguments": [ + { + "name": "keepref", + "type": "pure-token", + "token": "KEEPREF" + }, + { + "name": "delref", + "type": "pure-token", + "token": "DELREF" + }, + { + "name": "acked", + "type": "pure-token", + "token": "ACKED" + } + ] + }, { "name": "trim", "type": "block", diff --git a/src/commands/xdelex.json b/src/commands/xdelex.json new file mode 100644 index 000000000..cdb886e86 --- /dev/null +++ b/src/commands/xdelex.json @@ -0,0 +1,104 @@ +{ + "XDELEX": { + "summary": "Deletes one or multiple entries from the stream.", + "complexity": "O(1) for each single item to delete in the stream, regardless of the stream size.", + "group": "stream", + "since": "8.2.0", + "arity": -5, + "function": "xdelexCommand", + "command_flags": [ + "WRITE", + "FAST" + ], + "acl_categories": [ + "STREAM" + ], + "key_specs": [ + { + "flags": [ + "RW", + "DELETE" + ], + "begin_search": { + "index": { + "pos": 1 + } + }, + "find_keys": { + "range": { + "lastkey": 0, + "step": 1, + "limit": 0 + } + } + } + ], + "arguments": [ + { + "name": "key", + "type": "key", + "key_spec_index": 0 + }, + { + "name": "condition", + "type": "oneof", + "optional": true, + "arguments": [ + { + "name": "keepref", + "type": "pure-token", + "token": "KEEPREF" + }, + { + "name": "delref", + "type": "pure-token", + "token": "DELREF" + }, + { + "name": "acked", + "type": "pure-token", + "token": "ACKED" + } + ] + }, + { + "name": "ids", + "token": "IDS", + "type": "block", + "arguments": [ + { + "name": "numids", + "type": "integer" + }, + { + "name": "id", + "type": "string", + "multiple": true + } + ] + } + ], + "reply_schema": { + "description": "Array of results. Returns an array with -1 for each requested ID if the key does not exist.", + "type": "array", + "minItems": 0, + "maxItems": 4294967295, + "items": { + "oneOf": [ + { + "description": "The id does not exist in the provided stream key.", + "const": -1 + }, + { + "description": "Entry was deleted from the stream.", + "const": 1 + }, + { + "description": "Entry was not deleted, but there are still dangling references.", + "const": 2 + } + ] + } + } + } +} diff --git a/src/commands/xtrim.json b/src/commands/xtrim.json index 0b79cd468..bd962d5c8 100644 --- a/src/commands/xtrim.json +++ b/src/commands/xtrim.json @@ -10,6 +10,10 @@ [ "6.2.0", "Added the `MINID` trimming strategy and the `LIMIT` option." + ], + [ + "8.2.0", + "Added the `KEEPREF`, `DELREF` and `ACKED` options." ] ], "command_flags": [ @@ -95,6 +99,28 @@ "type": "integer", "optional": true, "since": "6.2.0" + }, + { + "name": "condition", + "type": "oneof", + "optional": true, + "arguments": [ + { + "name": "keepref", + "type": "pure-token", + "token": "KEEPREF" + }, + { + "name": "delref", + "type": "pure-token", + "token": "DELREF" + }, + { + "name": "acked", + "type": "pure-token", + "token": "ACKED" + } + ] } ] } diff --git a/src/rdb.c b/src/rdb.c index ce7b9efed..ea9f5e934 100644 --- a/src/rdb.c +++ b/src/rdb.c @@ -2973,6 +2973,7 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error) streamNACK *nack = streamCreateNACK(NULL); nack->delivery_time = rdbLoadMillisecondTime(rdb,RDB_VERSION); nack->delivery_count = rdbLoadLen(rdb,NULL); + nack->cgroup_ref_node = streamLinkCGroupToEntry(s, cgroup, rawid); if (rioGetReadError(rdb)) { rdbReportReadError("Stream PEL NACK loading failed."); decrRefCount(o); diff --git a/src/server.h b/src/server.h index 2824ee0d7..ee75e7877 100644 --- a/src/server.h +++ b/src/server.h @@ -4115,11 +4115,13 @@ void xreadCommand(client *c); void xgroupCommand(client *c); void xsetidCommand(client *c); void xackCommand(client *c); +void xackdelCommand(client *c); void xpendingCommand(client *c); void xclaimCommand(client *c); void xautoclaimCommand(client *c); void xinfoCommand(client *c); void xdelCommand(client *c); +void xdelexCommand(client *c); void xtrimCommand(client *c); void lolwutCommand(client *c); void aclCommand(client *c); diff --git a/src/stream.h b/src/stream.h index 146be3b12..3445c290a 100644 --- a/src/stream.h +++ b/src/stream.h @@ -21,6 +21,9 @@ typedef struct stream { streamID max_deleted_entry_id; /* The maximal ID that was deleted. */ uint64_t entries_added; /* All time count of elements added. */ rax *cgroups; /* Consumer groups dictionary: name -> streamCG */ + rax *cgroups_ref; /* Index mapping message IDs to their consumer groups. */ + streamID min_cgroup_last_id; /* The minimum ID of consume group. */ + unsigned int min_cgroup_last_id_valid: 1; } stream; /* We define an iterator to iterate stream items in an abstract way, without @@ -94,6 +97,7 @@ typedef struct streamNACK { uint64_t delivery_count; /* Number of times this message was delivered.*/ streamConsumer *consumer; /* The consumer this message was delivered to in the last delivery. */ + listNode *cgroup_ref_node; /* Reference to this NACK in the cgroups_ref list. */ } streamNACK; /* Stream propagation information, passed to functions in order to propagate @@ -144,4 +148,6 @@ long long streamEstimateDistanceFromFirstEverEntry(stream *s, streamID *id); int64_t streamTrimByLength(stream *s, long long maxlen, int approx); int64_t streamTrimByID(stream *s, streamID minid, int approx); +listNode *streamLinkCGroupToEntry(stream *s, streamCG *cg, unsigned char *key); + #endif diff --git a/src/t_stream.c b/src/t_stream.c index 83ba6ba38..e9b52fd96 100644 --- a/src/t_stream.c +++ b/src/t_stream.c @@ -33,13 +33,16 @@ * will return NULL. */ #define STREAM_LISTPACK_MAX_SIZE (1<<30) -void streamFreeCG(streamCG *cg); void streamFreeCGGeneric(void *cg); void streamFreeNACK(streamNACK *na); size_t streamReplyWithRangeFromConsumerPEL(client *c, stream *s, streamID *start, streamID *end, size_t count, streamConsumer *consumer); int streamParseStrictIDOrReply(client *c, robj *o, streamID *id, uint64_t missing_seq, int *seq_given); int streamParseIDOrReply(client *c, robj *o, streamID *id, uint64_t missing_seq); +int streamEntryIsReferenced(stream *s, streamID *id); +void streamCleanupEntryCGroupRefs(stream *s, streamID *id); +void streamUpdateCGroupLastId(stream *s, streamCG *cg, streamID *id); + /* ----------------------------------------------------------------------- * Low level stream encoding: a radix tree of listpacks. * ----------------------------------------------------------------------- */ @@ -57,6 +60,10 @@ stream *streamNew(void) { s->max_deleted_entry_id.ms = 0; s->entries_added = 0; s->cgroups = NULL; /* Created on demand to save memory when not used. */ + s->cgroups_ref = NULL; + s->min_cgroup_last_id.ms = UINT64_MAX; + s->min_cgroup_last_id.seq = UINT64_MAX; + s->min_cgroup_last_id_valid = 0; return s; } @@ -65,6 +72,8 @@ void freeStream(stream *s) { raxFreeWithCallback(s->rax, lpFreeGeneric); if (s->cgroups) raxFreeWithCallback(s->cgroups, streamFreeCGGeneric); + if (s->cgroups_ref) + raxFreeWithCallback(s->cgroups_ref, listReleaseGeneric); zfree(s); } @@ -199,6 +208,7 @@ robj *streamDup(robj *o) { streamNACK *new_nack = streamCreateNACK(NULL); new_nack->delivery_time = nack->delivery_time; new_nack->delivery_count = nack->delivery_count; + new_nack->cgroup_ref_node = streamLinkCGroupToEntry(new_s, new_cg, ri_cg_pel.key); raxInsert(new_cg->pel, ri_cg_pel.key, sizeof(streamID), new_nack, NULL); } raxStop(&ri_cg_pel); @@ -662,6 +672,7 @@ typedef struct { /* XADD + XTRIM common options */ int trim_strategy; /* TRIM_STRATEGY_* */ int trim_strategy_arg_idx; /* Index of the count in MAXLEN/MINID, for rewriting. */ + int delete_strategy; /* DELETE_STRATEGY_* */ int approx_trim; /* If 1 only delete whole radix tree nodes, so * the trim argument is not applied verbatim. */ long long limit; /* Maximum amount of entries to trim. If 0, no limitation @@ -676,6 +687,17 @@ typedef struct { #define TRIM_STRATEGY_MAXLEN 1 #define TRIM_STRATEGY_MINID 2 +typedef struct { + int startidx; /* Starting index of IDs in argv */ + long numids; /* Number of IDs to process */ + int delete_strategy; /* DELETE_STRATEGY_* */ +} streamAckDelArgs; + +#define DELETE_STRATEGY_NONE 0 +#define DELETE_STRATEGY_KEEPREF 1 /* Delete and keep references */ +#define DELETE_STRATEGY_DELREF 2 /* Delete from pending entries list */ +#define DELETE_STRATEGY_ACKED 3 /* Only delete messages that are acknowledged */ + /* Trim the stream 's' according to args->trim_strategy, and return the * number of elements removed from the stream. The 'approx' option, if non-zero, * specifies that the trimming must be performed in a approximated way in @@ -705,6 +727,7 @@ int64_t streamTrim(stream *s, streamAddTrimArgs *args) { int approx = args->approx_trim; int64_t limit = args->limit; int trim_strategy = args->trim_strategy; + int delete_strategy = args->delete_strategy; if (trim_strategy == TRIM_STRATEGY_NONE) return 0; @@ -725,21 +748,27 @@ int64_t streamTrim(stream *s, streamAddTrimArgs *args) { if (limit && (deleted + entries) > limit) break; - /* Check if we can remove the whole node. */ - int remove_node; - streamID master_id = {0}; /* For MINID */ + /* Check if we can remove the whole node */ + int remove_node = 0; /* Final decision flag for node removal */ + int node_eligible_for_remove = 0; /* Whether node meets the basic criteria for removal */ + streamID master_id = {0}; + /* Read the master ID from the radix tree key. */ + streamDecodeID(ri.key, &master_id); if (trim_strategy == TRIM_STRATEGY_MAXLEN) { - remove_node = s->length - entries >= maxlen; + node_eligible_for_remove = s->length - entries >= maxlen; } else { - /* Read the master ID from the radix tree key. */ - streamDecodeID(ri.key, &master_id); - /* Read last ID. */ streamID last_id = {0,0}; lpGetEdgeStreamID(lp, 0, &master_id, &last_id); /* We can remove the entire node id its last ID < 'id' */ - remove_node = streamCompareID(&last_id, id) < 0; + node_eligible_for_remove = streamCompareID(&last_id, id) < 0; + } + + if (node_eligible_for_remove && delete_strategy == DELETE_STRATEGY_KEEPREF) { + /* With KEEPREF strategy, we can remove the whole node directly since we don't need + * to check or clean up consumer group references. */ + remove_node = 1; } if (remove_node) { @@ -785,11 +814,9 @@ int64_t streamTrim(stream *s, streamAddTrimArgs *args) { int64_t seq_delta = lpGetInteger(p); p = lpNext(lp, p); /* Skip ID seq delta */ - streamID currid = {0}; /* For MINID */ - if (trim_strategy == TRIM_STRATEGY_MINID) { - currid.ms = master_id.ms + ms_delta; - currid.seq = master_id.seq + seq_delta; - } + streamID currid = {0}; + currid.ms = master_id.ms + ms_delta; + currid.seq = master_id.seq + seq_delta; int stop; if (trim_strategy == TRIM_STRATEGY_MAXLEN) { @@ -813,17 +840,38 @@ int64_t streamTrim(stream *s, streamAddTrimArgs *args) { while(to_skip--) p = lpNext(lp,p); /* Skip the whole entry. */ p = lpNext(lp,p); /* Skip the final lp-count field. */ - /* Mark the entry as deleted. */ + /* Mark the entry as deleted if allowed. */ if (!(flags & STREAM_ITEM_FLAG_DELETED)) { - intptr_t delta = p - lp; - flags |= STREAM_ITEM_FLAG_DELETED; - lp = lpReplaceInteger(lp, &pcopy, flags); - deleted_from_lp++; - s->length--; - p = lp + delta; + int can_delete = 1; + if (delete_strategy == DELETE_STRATEGY_ACKED) { + /* Only delete entry that has been acknowledged by all consumer groups. */ + can_delete = (streamEntryIsReferenced(s, &currid) == 0); + } else if (delete_strategy == DELETE_STRATEGY_DELREF) { + /* Remove all consumer group references for this entry */ + streamCleanupEntryCGroupRefs(s, &currid); + } + + if (can_delete) { + /* Mark the entry as deleted. */ + intptr_t delta = p - lp; + flags |= STREAM_ITEM_FLAG_DELETED; + lp = lpReplaceInteger(lp, &pcopy, flags); + deleted_from_lp++; + s->length--; + p = lp + delta; + } } } deleted += deleted_from_lp; + /* If this node was originally eligible for removal but we couldn't remove it upfront + * due to delete strategy constraints, and now we've processed and deleted all entries + * in the node, we can finally remove the entire node. */ + if (node_eligible_for_remove && deleted_from_lp == entries) { + lpFree(lp); + raxRemove(s->rax,ri.key,ri.key_len,NULL); + raxSeek(&ri,">=",ri.key,ri.key_len); + continue; + } /* Now we update the entries/deleted counters. */ p = lpFirst(lp); @@ -844,6 +892,12 @@ int64_t streamTrim(stream *s, streamAddTrimArgs *args) { /* Update the listpack with the new pointer. */ raxInsert(s->rax,ri.key,ri.key_len,lp,NULL); + /* If the node is eligible for removal but we couldn't remove it due to delete strategy + * constraints (we need to check each entry individually), continue to the next node + * instead of stopping here. */ + if (node_eligible_for_remove) + continue; + break; /* If we are here, there was enough to delete in the current node, so no need to go to the next node. */ } @@ -866,7 +920,8 @@ int64_t streamTrimByLength(stream *s, long long maxlen, int approx) { .trim_strategy = TRIM_STRATEGY_MAXLEN, .approx_trim = approx, .limit = approx ? 100 * server.stream_node_max_entries : 0, - .maxlen = maxlen + .maxlen = maxlen, + .delete_strategy = DELETE_STRATEGY_KEEPREF }; return streamTrim(s, &args); } @@ -877,7 +932,8 @@ int64_t streamTrimByID(stream *s, streamID minid, int approx) { .trim_strategy = TRIM_STRATEGY_MINID, .approx_trim = approx, .limit = approx ? 100 * server.stream_node_max_entries : 0, - .minid = minid + .minid = minid, + .delete_strategy = DELETE_STRATEGY_KEEPREF }; return streamTrim(s, &args); } @@ -891,6 +947,7 @@ int64_t streamTrimByID(stream *s, streamID minid, int approx) { static int streamParseAddOrTrimArgsOrReply(client *c, streamAddTrimArgs *args, int xadd) { /* Initialize arguments to defaults */ memset(args, 0, sizeof(*args)); + args->delete_strategy = DELETE_STRATEGY_NONE; /* Parse options. */ int i = 2; /* This is the first argument position where we could @@ -964,6 +1021,12 @@ static int streamParseAddOrTrimArgsOrReply(client *c, streamAddTrimArgs *args, i } limit_given = 1; i++; + } else if (!strcasecmp(opt,"keepref") && args->delete_strategy == DELETE_STRATEGY_NONE) { + args->delete_strategy = DELETE_STRATEGY_KEEPREF; + } else if (!strcasecmp(opt,"delref") && args->delete_strategy == DELETE_STRATEGY_NONE) { + args->delete_strategy = DELETE_STRATEGY_DELREF; + } else if (!strcasecmp(opt,"acked") && args->delete_strategy == DELETE_STRATEGY_NONE) { + args->delete_strategy = DELETE_STRATEGY_ACKED; } else if (xadd && !strcasecmp(opt,"nomkstream")) { args->no_mkstream = 1; } else if (xadd) { @@ -1019,9 +1082,66 @@ static int streamParseAddOrTrimArgsOrReply(client *c, streamAddTrimArgs *args, i } } + /* Set default consumer group reference handling to KEEPREF if none was specified */ + if (args->delete_strategy == DELETE_STRATEGY_NONE) + args->delete_strategy = DELETE_STRATEGY_KEEPREF; + return i; } +static int streamParseAckDelArgsOrReply(client *c, int start_pos, streamAckDelArgs *args) { + /* Initialize arguments to defaults */ + memset(args, 0, sizeof(*args)); + args->startidx = -1; + args->delete_strategy = DELETE_STRATEGY_NONE; + + /* Parse command options */ + int j = start_pos; + while (j < c->argc) { + char *opt = c->argv[j]->ptr; + if (!strcasecmp(opt, "KEEPREF") && args->delete_strategy == DELETE_STRATEGY_NONE) { + args->delete_strategy = DELETE_STRATEGY_KEEPREF; + j++; + } else if (!strcasecmp(opt, "DELREF") && args->delete_strategy == DELETE_STRATEGY_NONE) { + args->delete_strategy = DELETE_STRATEGY_DELREF; + j++; + } else if (!strcasecmp(opt, "ACKED") && args->delete_strategy == DELETE_STRATEGY_NONE) { + args->delete_strategy = DELETE_STRATEGY_ACKED; + j++; + } else if (!strcasecmp(opt, "IDS") && j+1 < c->argc) { + /* Parse the number of IDs */ + if (getRangeLongFromObjectOrReply(c, c->argv[j+1], 1, LONG_MAX, + &args->numids, "Number of IDs must be a positive integer") != C_OK) + { + return 0; + } + + /* Verify that the specified number of IDs matches the actual arguments */ + if (args->numids > (c->argc - j - 2)) { + addReplyError(c, "The `numids` parameter must match the number of arguments"); + return 0; + } + + args->startidx = j + 2; /* Skip "IDS" and numids */ + j = args->startidx + args->numids; + } else { + addReplyErrorObject(c,shared.syntaxerr); + return 0; + } + } + + if (args->startidx == -1) { + addReplyError(c, "IDS option is required"); + return 0; + } + + /* Set default consumer group reference handling to KEEPREF if none was specified */ + if (args->delete_strategy == DELETE_STRATEGY_NONE) + args->delete_strategy = DELETE_STRATEGY_KEEPREF; + + return 1; +} + /* Initialize the stream iterator, so that we can call iterating functions * to get the next items. This requires a corresponding streamIteratorStop() * at the end. The 'rev' parameter controls the direction. If it's zero the @@ -1708,7 +1828,7 @@ size_t streamReplyWithRange(client *c, stream *s, streamID *start, streamID *end /* The group's counter may be invalid, so we try to obtain it. */ group->entries_read = streamEstimateDistanceFromFirstEverEntry(s,&id); } - group->last_id = id; + streamUpdateCGroupLastId(s, group, &id); /* 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, @@ -1770,6 +1890,8 @@ size_t streamReplyWithRange(client *c, stream *s, streamID *start, streamID *end nack->delivery_count = 1; /* Add the entry in the new consumer local PEL. */ raxInsert(consumer->pel,buf,sizeof(buf),nack,NULL); + } else if (group_inserted == 1 && consumer_inserted == 1) { + nack->cgroup_ref_node = streamLinkCGroupToEntry(s, group, buf); } else if (group_inserted == 1 && consumer_inserted == 0) { serverPanic("NACK half-created. Should not be possible."); } @@ -2467,6 +2589,127 @@ cleanup: /* Cleanup. */ * Low level implementation of consumer groups * ----------------------------------------------------------------------- */ +/* Update a consumer group's last_id and handle minimum last_id tracking. + * we will recalculate the minimum last_id when needed. */ +void streamUpdateCGroupLastId(stream *s, streamCG *cg, streamID *id) { + /* When a consumer group's last_id is updated, we need to invalidate the cached + * minimum last_id in two cases: + * 1. If the consumer group's previous last_id equals the minimum last_id. + * 2. If the new ID being set is smaller than the current minimum last_id. */ + if (s->min_cgroup_last_id_valid && + (streamCompareID(&cg->last_id, &s->min_cgroup_last_id) == 0 || + streamCompareID(id, &s->min_cgroup_last_id) < 0)) + { + s->min_cgroup_last_id_valid = 0; + } + cg->last_id = *id; +} + +/* Link a consumer group to a stream entry in the cgroups_ref index. + * Returns a pointer to the list node, so that it can be used for future deletion. */ +listNode *streamLinkCGroupToEntry(stream *s, streamCG *cg, unsigned char *key) { + list *cglist; + + if (!s->cgroups_ref) + s->cgroups_ref = raxNew(); + + /* Try to find the list for this stream ID, create it if it doesn't exist */ + if (!raxFind(s->cgroups_ref, key, sizeof(streamID), (void**)&cglist)) { + cglist = listCreate(); + serverAssert(raxInsert(s->cgroups_ref, key, sizeof(streamID), cglist, NULL)); + } + + /* Add the consumer group to the list and return the list node */ + listAddNodeTail(cglist, cg); + return listLast(cglist); +} + +/* Unlink a consumer group reference from the entry index for a specific stream ID. + * This is called when a message is acknowledged or when a consumer group is deleted. */ +void streamUnlinkEntryFromCGroupRef(stream *s, streamNACK *na, unsigned char *key) { + list *cglist; + if (!s->cgroups_ref) return; + if (raxFind(s->cgroups_ref, key, sizeof(streamID), (void**)&cglist)) { + listDelNode(cglist, na->cgroup_ref_node); + + /* If the list is now empty, remove it from the index. */ + if (listLength(cglist) == 0) { + raxRemove(s->cgroups_ref, key, sizeof(streamID), NULL); + listRelease(cglist); + } + } +} + +/* Remove all consumer group references to a specific stream message. */ +void streamCleanupEntryCGroupRefs(stream *s, streamID *id) { + if (!s->cgroups_ref) return; + list *cglist; + listIter li; + listNode *ln; + unsigned char buf[sizeof(streamID)]; + streamEncodeID(buf, id); + + /* If message is not in any consumer group, nothing to do */ + if (!raxFind(s->cgroups_ref, buf, sizeof(streamID), (void **)&cglist)) + return; + + listRewind(cglist, &li); + while ((ln = listNext(&li))) { + streamNACK *nack; + streamCG *group = listNodeValue(ln); + + /* Find the message in this consumer group's PEL */ + serverAssert(raxFind(group->pel, buf, sizeof(buf), (void **)&nack)); + + /* Remove from group and consumer PELs */ + raxRemove(group->pel, buf, sizeof(buf), NULL); + raxRemove(nack->consumer->pel, buf, sizeof(buf), NULL); + /* Since we're removing all references from the cgroups_ref, we can directly + * free the NACK without unlinking it from the cgroups_ref. */ + streamFreeNACK(nack); + } + + raxRemove(s->cgroups_ref, buf, sizeof(streamID), NULL); + listRelease(cglist); +} + +/* Check if a stream entry is still referenced by any consumer group. + * + * An entry is considered referenced if: + * 1. Its ID is smaller than the minimum last_id of all consumer groups, + * which means at least one group hasn't read it yet. + * 2. It exists in any consumer group's PEL. + * + * Returns 1 if the entry is referenced, 0 if it's fully acknowledged by all groups. */ +int streamEntryIsReferenced(stream *s, streamID *id) { + if (!s->cgroups || !raxSize(s->cgroups)) return 0; + if (!s->min_cgroup_last_id_valid) { + /* If the cached minimum last_id is invalid, we need to recalculate it + * by iterating through all consumer groups to find the minimum last_id */ + s->min_cgroup_last_id_valid = 1; + s->min_cgroup_last_id.ms = UINT64_MAX; + s->min_cgroup_last_id.seq = UINT64_MAX; + raxIterator ri; + raxStart(&ri, s->cgroups); + raxSeek(&ri, "^", NULL, 0); + while (raxNext(&ri)) { + streamCG *cg = ri.data; + if (streamCompareID(&cg->last_id, &s->min_cgroup_last_id) < 0) + s->min_cgroup_last_id = cg->last_id; + } + raxStop(&ri); + } + + /* The consume group doesn't read it. */ + if (streamCompareID(&s->min_cgroup_last_id, id) < 0) + return 1; + + /* Check if the message is in any consumer group's PEL */ + unsigned char buf[sizeof(streamID)]; + streamEncodeID(buf, id); + return raxFind(s->cgroups_ref, buf, sizeof(streamID), NULL); +} + /* Create a NACK entry setting the delivery count to 1 and the delivery * time to the current time. The NACK consumer will be set to the one * specified as argument of the function. */ @@ -2475,6 +2718,7 @@ streamNACK *streamCreateNACK(streamConsumer *consumer) { nack->delivery_time = commandTimeSnapshot(); nack->delivery_count = 1; nack->consumer = consumer; + nack->cgroup_ref_node = NULL; /* Will be set when added to cgroups_ref */ return nack; } @@ -2483,6 +2727,13 @@ void streamFreeNACK(streamNACK *na) { zfree(na); } +/* Free a NACK entry and remove its reference from the cgroups_ref. + * This ensures proper cleanup of the consumer group list associated with the message ID. */ +void streamDestroyNACK(stream *s, streamNACK *na, unsigned char *key) { + streamUnlinkEntryFromCGroupRef(s, na, key); + zfree(na); +} + /* Generic version of streamFreeNACK. */ void streamFreeNACKGeneric(void *na) { streamFreeNACK((streamNACK *)na); @@ -2517,19 +2768,35 @@ streamCG *streamCreateCG(stream *s, char *name, size_t namelen, streamID *id, lo streamCG *cg = zmalloc(sizeof(*cg)); cg->pel = raxNew(); cg->consumers = raxNew(); - cg->last_id = *id; + cg->last_id.ms = 0; + cg->last_id.seq = 0; + streamUpdateCGroupLastId(s, cg, id); cg->entries_read = entries_read; raxInsert(s->cgroups,(unsigned char*)name,namelen,cg,NULL); return cg; } /* Free a consumer group and all its associated data. */ -void streamFreeCG(streamCG *cg) { +static void streamFreeCG(streamCG *cg) { raxFreeWithCallback(cg->pel, streamFreeNACKGeneric); raxFreeWithCallback(cg->consumers, streamFreeConsumerGeneric); zfree(cg); } +/* Destroy a consumer group and clean up all associated references. */ +void streamDestroyCG(stream *s, streamCG *cg) { + raxIterator it; + raxStart(&it, cg->pel); + raxSeek(&it, "^", NULL, 0); + while (raxNext(&it)) { + streamNACK *nack = it.data; + streamUnlinkEntryFromCGroupRef(s, nack, it.key); + } + raxStop(&it); + + streamFreeCG(cg); +} + /* Generic version of streamFreeCG. */ void streamFreeCGGeneric(void *cg) { streamFreeCG((streamCG *)cg); @@ -2577,7 +2844,7 @@ streamConsumer *streamLookupConsumer(streamCG *cg, sds name) { } /* Delete the consumer specified in the consumer group 'cg'. */ -void streamDelConsumer(streamCG *cg, streamConsumer *consumer) { +void streamDelConsumer(stream *s, streamCG *cg, streamConsumer *consumer) { /* Iterate all the consumer pending messages, deleting every corresponding * entry from the global entry. */ raxIterator ri; @@ -2585,8 +2852,9 @@ void streamDelConsumer(streamCG *cg, streamConsumer *consumer) { raxSeek(&ri,"^",NULL,0); while(raxNext(&ri)) { streamNACK *nack = ri.data; - raxRemove(cg->pel,ri.key,ri.key_len,NULL); + streamUnlinkEntryFromCGroupRef(s, nack, ri.key); streamFreeNACK(nack); + raxRemove(cg->pel,ri.key,ri.key_len,NULL); } raxStop(&ri); @@ -2728,7 +2996,7 @@ NULL } else if (streamParseIDOrReply(c,c->argv[4],&id,0) != C_OK) { return; } - cg->last_id = id; + streamUpdateCGroupLastId(s, cg, &id); cg->entries_read = entries_read; addReply(c,shared.ok); server.dirty++; @@ -2736,7 +3004,7 @@ NULL } else if (!strcasecmp(opt,"DESTROY") && c->argc == 4) { if (cg) { raxRemove(s->cgroups,(unsigned char*)grpname,sdslen(grpname),NULL); - streamFreeCG(cg); + streamDestroyCG(s, cg); addReply(c,shared.cone); server.dirty++; notifyKeyspaceEvent(NOTIFY_STREAM,"xgroup-destroy", @@ -2757,7 +3025,7 @@ NULL /* Delete the consumer and returns the number of pending messages * that were yet associated with such a consumer. */ pending = raxSize(consumer->pel); - streamDelConsumer(cg,consumer); + streamDelConsumer(s,cg,consumer); server.dirty++; notifyKeyspaceEvent(NOTIFY_STREAM,"xgroup-delconsumer", c->argv[2],c->db->id); @@ -2891,7 +3159,7 @@ void xackCommand(client *c) { streamNACK *nack = result; raxRemove(group->pel,buf,sizeof(buf),NULL); raxRemove(nack->consumer->pel,buf,sizeof(buf),NULL); - streamFreeNACK(nack); + streamDestroyNACK(kv->ptr, nack, buf); acknowledged++; server.dirty++; } @@ -2901,6 +3169,115 @@ cleanup: if (ids != static_ids) zfree(ids); } +/* Used by xackdelCommand() */ +typedef enum XAckDelRes { + XACKDEL_NO_ID = -1, /* ID not found in PEL. */ + XACKDEL_DELETED = 1, /* Message acknowledged and deleted. */ + XACKDEL_STILL_REFERENCED = 2, /* Message acknowledged but not deleted (still referenced). */ +} XAckDelRes; + +/* XACKDEL [KEEPREF|DELREF|ACKED] [IDS ] + * Acknowledges messages as processed and deletes them from the stream. + * + * Returns an array of status codes for each ID, indicating whether it + * was deleted, still referenced, or not found. */ +void xackdelCommand(client *c) { + stream *s = NULL; + streamCG *group = NULL; + kvobj *kv = lookupKeyRead(c->db, c->argv[1]); + if (checkType(c, kv, OBJ_STREAM)) return; /* Type error. */ + + /* Parse command options */ + streamAckDelArgs args; + if (!streamParseAckDelArgsOrReply(c, 3, &args)) return; + + /* Reply null if the key doesn't exist or the group doesn't exist.*/ + if (!kv || !(group = streamLookupCG(kv->ptr, c->argv[2]->ptr))) { + addReplyArrayLen(c, args.numids); + for (int i = 0; i < args.numids; i++) + addReplyLongLong(c, XACKDEL_NO_ID); + return; + } + + /* Start parsing the IDs, so that we abort ASAP if there is a syntax + * error: the return value of this command cannot be an error in case + * the client successfully acknowledged some messages, so it should be + * executed in a "all or nothing" fashion. */ + streamID static_ids[STREAMID_STATIC_VECTOR_LEN]; + streamID *ids = static_ids; + for (int j = 0; j < args.numids; j++) { + if (streamParseStrictIDOrReply(c,c->argv[j+args.startidx],&ids[j],0,NULL) != C_OK) + goto cleanup; + } + + s = kv->ptr; + int first_entry = 0; + int deleted = 0; + addReplyArrayLen(c, args.numids); + for (int j = 0; j < args.numids; j++) { + int res = XACKDEL_NO_ID; + streamID *id = &ids[j]; + unsigned char buf[sizeof(streamID)]; + streamEncodeID(buf,id); + + /* Lookup the ID in the group PEL: it will have a reference to the + * NACK structure that will have a reference to the consumer, so that + * we are able to remove the entry from both PELs. */ + void *result; + if (raxFind(group->pel,buf,sizeof(buf),&result)) { + streamNACK *nack = result; + raxRemove(group->pel,buf,sizeof(buf),NULL); + raxRemove(nack->consumer->pel,buf,sizeof(buf),NULL); + streamDestroyNACK(s, nack, buf); + server.dirty++; + + int can_delete = 1; + if (args.delete_strategy == DELETE_STRATEGY_ACKED) { + /* Only delete if acknowledged by all consumer groups */ + if (streamEntryIsReferenced(s, id)) + can_delete = 0; + } else if (args.delete_strategy == DELETE_STRATEGY_DELREF) { + streamCleanupEntryCGroupRefs(s, id); + } + + if (can_delete && streamDeleteItem(s,id)) { + /* We want to know if the first entry in the stream was deleted + * so we can later set the new one. */ + if (streamCompareID(id,&s->first_id) == 0) { + first_entry = 1; + } + /* Update the stream's maximal tombstone if needed. */ + if (streamCompareID(id,&s->max_deleted_entry_id) > 0) { + s->max_deleted_entry_id = *id; + } + deleted++; + } + + /* If the entry was in the PEL but not found in the stream, + * we still consider it successfully deleted. */ + res = can_delete ? XACKDEL_DELETED : XACKDEL_STILL_REFERENCED; + } + addReplyLongLong(c, res); + } + + /* Update the stream's first ID. */ + if (deleted) { + if (s->length == 0) { + s->first_id.ms = 0; + s->first_id.seq = 0; + } else if (first_entry) { + streamGetEdgeID(s,1,1,&s->first_id); + } + + /* Propagate the write. */ + signalModifiedKey(c,c->db,c->argv[1]); + notifyKeyspaceEvent(NOTIFY_STREAM,"xdel",c->argv[1],c->db->id); + } + +cleanup: + if (ids != static_ids) zfree(ids); +} + /* XPENDING [[IDLE ] []] * * If start and stop are omitted, the command just outputs information about @@ -3232,7 +3609,7 @@ void xclaimCommand(client *c) { } if (streamCompareID(&last_id,&group->last_id) > 0) { - group->last_id = last_id; + streamUpdateCGroupLastId(o->ptr, group, &last_id); propagate_last_id = 1; } @@ -3282,7 +3659,7 @@ void xclaimCommand(client *c) { /* Release the NACK */ raxRemove(group->pel,buf,sizeof(buf),NULL); raxRemove(nack->consumer->pel,buf,sizeof(buf),NULL); - streamFreeNACK(nack); + streamDestroyNACK(o->ptr, nack, buf); } continue; } @@ -3296,6 +3673,7 @@ void xclaimCommand(client *c) { /* Create the NACK. */ nack = streamCreateNACK(NULL); raxInsert(group->pel,buf,sizeof(buf),nack,NULL); + nack->cgroup_ref_node = streamLinkCGroupToEntry(o->ptr, group, buf); } if (nack != NULL) { @@ -3471,7 +3849,7 @@ void xautoclaimCommand(client *c) { /* Clear this entry from the PEL, it no longer exists */ raxRemove(group->pel,ri.key,ri.key_len,NULL); raxRemove(nack->consumer->pel,ri.key,ri.key_len,NULL); - streamFreeNACK(nack); + streamDestroyNACK(o->ptr, nack, ri.key); /* Remember the ID for later */ deleted_ids[deleted_id_num++] = id; raxSeek(&ri,">=",ri.key,ri.key_len); @@ -3608,6 +3986,107 @@ cleanup: if (ids != static_ids) zfree(ids); } +/* Used by xdelexCommand() */ +typedef enum XDelexRes { + XDELEX_NO_ID = -1, /* ID not found in the stream. */ + XDELEX_DELETED = 1, /* Message deleted. */ + XDELEX_STILL_REFERENCED = 2, /* Message not deleted (still referenced). */ +} XDelexRes; + +/* XDELEX [KEEPREF|DELREF|ACKED] [IDS ] + * + * Removes specified entries from the stream. Returns an array of status codes for + * each ID, indicating whether it was deleted, still referenced, or not found. */ +void xdelexCommand(client *c) { + kvobj *kv = lookupKeyWrite(c->db, c->argv[1]); + if (checkType(c, kv, OBJ_STREAM)) return; + + /* Parse command options */ + streamAckDelArgs args; + if (!streamParseAckDelArgsOrReply(c, 2, &args)) return; + + /* Non-existing keys and empty stream are the same thing. Reply null if the + * key does not exist.*/ + if (!kv) { + addReplyArrayLen(c, args.numids); + for (int i = 0; i < args.numids; i++) + addReplyLongLong(c, XDELEX_NO_ID); + return; + } + + /* We need to sanity check the IDs passed to start. Even if not + * a big issue, it is not great that the command is only partially + * executed because at some point an invalid ID is parsed. */ + streamID static_ids[STREAMID_STATIC_VECTOR_LEN]; + streamID *ids = static_ids; + if (args.numids > STREAMID_STATIC_VECTOR_LEN) + ids = zmalloc(sizeof(streamID)*args.numids); + for (int j = 0; j < args.numids; j++) { + if (streamParseStrictIDOrReply(c,c->argv[j+args.startidx],&ids[j],0,NULL) != C_OK) + goto cleanup; + } + + stream *s = kv->ptr; + int first_entry = 0; + int deleted = 0; + addReplyArrayLen(c, args.numids); + for (int j = 0; j < args.numids; j++) { + int res = XDELEX_NO_ID; + streamID *id = &ids[j]; + unsigned char buf[sizeof(streamID)]; + streamEncodeID(buf,id); + + int can_delete = 1; + if (args.delete_strategy == DELETE_STRATEGY_ACKED) { + /* Only delete if acknowledged by all consumer groups */ + if (streamEntryIsReferenced(s, id)) + can_delete = 0; + } else if (args.delete_strategy == DELETE_STRATEGY_DELREF) { + streamCleanupEntryCGroupRefs(s, id); + } + + if (can_delete) { /* can_delete being true doesn't guarantee the ID exists */ + if (streamDeleteItem(s,id)) { + /* We want to know if the first entry in the stream was deleted + * so we can later set the new one. */ + if (streamCompareID(id,&s->first_id) == 0) { + first_entry = 1; + } + /* Update the stream's maximal tombstone if needed. */ + if (streamCompareID(id,&s->max_deleted_entry_id) > 0) { + s->max_deleted_entry_id = *id; + } + deleted++; + res = XDELEX_DELETED; + } else { + /* This id doesn't exist. */ + } + } else { + res = XDELEX_STILL_REFERENCED; + } + + addReplyLongLong(c, res); + } + + /* Update the stream's first ID. */ + if (deleted) { + if (s->length == 0) { + s->first_id.ms = 0; + s->first_id.seq = 0; + } else if (first_entry) { + streamGetEdgeID(s,1,1,&s->first_id); + } + + /* Propagate the write. */ + signalModifiedKey(c,c->db,c->argv[1]); + notifyKeyspaceEvent(NOTIFY_STREAM,"xdel",c->argv[1],c->db->id); + server.dirty += deleted; + } + +cleanup: + if (ids != static_ids) zfree(ids); +} + /* General form: XTRIM [... options ...] * * List of options: @@ -3623,6 +4102,12 @@ cleanup: * count in order to demand approximated trimming * (like XADD MINID option). * + * Consumer group reference handling (optional, defaults to KEEPREF): + * + * KEEPREF -- Keeps existing consumer group references + * DELREF -- Clean up all consumer group references + * ACKED -- Only delete messages that are acknowledged + * * Other options: * * LIMIT -- The maximum number of entries to trim. diff --git a/tests/modules/cmdintrospection.c b/tests/modules/cmdintrospection.c index 1a5e4863b..188207542 100644 --- a/tests/modules/cmdintrospection.c +++ b/tests/modules/cmdintrospection.c @@ -34,6 +34,7 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) * XADD). */ {"6.2.0", "Added the `NOMKSTREAM` option, `MINID` trimming strategy and the `LIMIT` option."}, {"7.0.0", "Added support for the `-*` explicit ID form."}, + {"8.2.0", "Added the `KEEPREF`, `DELREF` and `ACKED` options."}, {0} }, .key_specs = (RedisModuleCommandKeySpec[]){ @@ -60,6 +61,29 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) .since = "6.2.0", .flags = REDISMODULE_CMD_ARG_OPTIONAL }, + { + .name = "condition", + .type = REDISMODULE_ARG_TYPE_ONEOF, + .flags = REDISMODULE_CMD_ARG_OPTIONAL, + .subargs = (RedisModuleCommandArg[]){ + { + .name = "keepref", + .type = REDISMODULE_ARG_TYPE_PURE_TOKEN, + .token = "KEEPREF" + }, + { + .name = "delref", + .type = REDISMODULE_ARG_TYPE_PURE_TOKEN, + .token = "DELREF" + }, + { + .name = "acked", + .type = REDISMODULE_ARG_TYPE_PURE_TOKEN, + .token = "ACKED" + }, + {0} + } + }, { .name = "trim", .type = REDISMODULE_ARG_TYPE_BLOCK, diff --git a/tests/support/util.tcl b/tests/support/util.tcl index b62d9301f..b8b917d2b 100644 --- a/tests/support/util.tcl +++ b/tests/support/util.tcl @@ -741,7 +741,7 @@ proc generate_fuzzy_traffic_on_key {key type duration} { set zset_commands {ZADD ZCARD ZCOUNT ZINCRBY ZINTERSTORE ZLEXCOUNT ZPOPMAX ZPOPMIN ZRANGE ZRANGEBYLEX ZRANGEBYSCORE ZRANK ZREM ZREMRANGEBYLEX ZREMRANGEBYRANK ZREMRANGEBYSCORE ZREVRANGE ZREVRANGEBYLEX ZREVRANGEBYSCORE ZREVRANK ZSCAN ZSCORE ZUNIONSTORE ZRANDMEMBER} set list_commands {LINDEX LINSERT LLEN LPOP LPOS LPUSH LPUSHX LRANGE LREM LSET LTRIM RPOP RPOPLPUSH RPUSH RPUSHX} set set_commands {SADD SCARD SDIFF SDIFFSTORE SINTER SINTERSTORE SISMEMBER SMEMBERS SMOVE SPOP SRANDMEMBER SREM SSCAN SUNION SUNIONSTORE} - set stream_commands {XACK XADD XCLAIM XDEL XGROUP XINFO XLEN XPENDING XRANGE XREAD XREADGROUP XREVRANGE XTRIM} + set stream_commands {XACK XADD XCLAIM XDEL XGROUP XINFO XLEN XPENDING XRANGE XREAD XREADGROUP XREVRANGE XTRIM XDELEX XACKDEL} set vset_commands {VADD VREM} set commands [dict create string $string_commands hash $hash_commands zset $zset_commands list $list_commands set $set_commands stream $stream_commands vectorset $vset_commands] diff --git a/tests/unit/type/stream-cgroups.tcl b/tests/unit/type/stream-cgroups.tcl index 8bc92c09c..2fbe794bf 100644 --- a/tests/unit/type/stream-cgroups.tcl +++ b/tests/unit/type/stream-cgroups.tcl @@ -548,6 +548,25 @@ start_server { assert_equal [s total_error_replies] 1 } + test {XGROUP DESTROY removes all consumer group references} { + r DEL mystream + for {set j 0} {$j < 5} {incr j} { + r XADD mystream $j-1 item $j + } + + r XGROUP CREATE mystream mygroup 0 + r XREADGROUP GROUP mygroup consumer1 STREAMS mystream > + assert {[lindex [r XPENDING mystream mygroup] 0] == 5} + + # Try to delete a message with ACKED - should fail because both groups have references + assert_equal {2 2 2 2 2} [r XDELEX mystream ACKED IDS 5 0-1 1-1 2-1 3-1 4-1] + + # Destroy one consumer group, and then we can delete all the entries with ACKED. + r XGROUP DESTROY mystream mygroup + assert_equal {1 1 1 1 1} [r XDELEX mystream ACKED IDS 5 0-1 1-1 2-1 3-1 4-1] + assert_equal 0 [r XLEN mystream] + } + test {RENAME can unblock XREADGROUP with data} { r del mystream{t} r XGROUP CREATE mystream{t} mygroup $ MKSTREAM @@ -1505,4 +1524,111 @@ start_server { assert_equal [r xinfo groups mystream] $grpinfo } } + + start_server {} { + test "XACKDEL wrong number of args" { + assert_error {*wrong number of arguments for 'xackdel' command} {r XACKDEL} + assert_error {*wrong number of arguments for 'xackdel' command} {r XACKDEL s} + assert_error {*wrong number of arguments for 'xackdel' command} {r XACKDEL s g} + } + + test "XACKDEL should return empty array when key doesn't exist or group doesn't exist" { + r DEL s + assert_equal {-1 -1} [r XACKDEL s g IDS 2 1-1 2-2] ;# the key doesn't exist + + r XADD s 1-0 f v + assert_equal {-1 -1} [r XACKDEL s g IDS 2 1-1 2-2] ;# the key exists but the group doesn't exist + } + + test "XACKDEL IDS parameter validation" { + r DEL s + r XADD s 1-0 f v + r XGROUP CREATE s g 0 + + # Test invalid numids + assert_error {*Number of IDs must be a positive integer*} {r XACKDEL s g IDS abc 1-1} + assert_error {*Number of IDs must be a positive integer*} {r XACKDEL s g IDS 0 1-1} + assert_error {*Number of IDs must be a positive integer*} {r XACKDEL s g IDS -5 1-1} + + # Test whether numids is equal to the number of IDs provided + assert_error {*The `numids` parameter must match the number of arguments*} {r XACKDEL s g IDS 3 1-1 2-2} + assert_error {*syntax error*} {r XACKDEL s g IDS 1 1-1 2-2} + } + + test "XACKDEL KEEPREF/DELREF/ACKED parameter validation" { + # Test mutually exclusive options + assert_error {*syntax error*} {r XACKDEL s g KEEPREF DELREF IDS 1 1-1} + assert_error {*syntax error*} {r XACKDEL s g KEEPREF ACKED IDS 1 1-1} + assert_error {*syntax error*} {r XACKDEL s g DELREF ACKED IDS 1 1-1} + } + + test "XACKDEL with DELREF option acknowledges will remove entry from all PELs" { + r DEL mystream + r XADD mystream 1-0 f v + r XADD mystream 2-0 f v + + # Create two consumer groups + r XGROUP CREATE mystream group1 0 + r XGROUP CREATE mystream group2 0 + r XREADGROUP GROUP group1 consumer1 STREAMS mystream > + r XREADGROUP GROUP group2 consumer2 STREAMS mystream > + + # Verify the message was removed from both groups' PELs when with DELREF + assert_equal {1 1} [r XACKDEL mystream group1 DELREF IDS 2 1-0 2-0] + assert_equal 0 [r XLEN mystream] + assert_equal {0 {} {} {}} [r XPENDING mystream group1] + assert_equal {0 {} {} {}} [r XPENDING mystream group2] + assert_equal {-1 -1} [r XACKDEL mystream group2 DELREF IDS 2 1-0 2-0] + } + + test "XACKDEL with ACKED option only deletes messages acknowledged by all groups" { + r DEL mystream + r XADD mystream 1-0 f v + r XADD mystream 2-0 f v + + # Create two consumer groups + r XGROUP CREATE mystream group1 0 + r XGROUP CREATE mystream group2 0 + r XREADGROUP GROUP group1 consumer1 STREAMS mystream > + r XREADGROUP GROUP group2 consumer2 STREAMS mystream > + + # The message is referenced by two groups. + # Even after one of them is ack, it still can't be deleted. + assert_equal {2 2} [r XACKDEL mystream group1 ACKED IDS 2 1-0 2-0] + assert_equal 2 [r XLEN mystream] + assert_equal {0 {} {} {}} [r XPENDING mystream group1] + assert_equal {2 1-0 2-0 {{consumer2 2}}} [r XPENDING mystream group2] + + # When these messages are dereferenced by all groups, they can be deleted. + assert_equal {1 1} [r XACKDEL mystream group2 ACKED IDS 2 1-0 2-0] + assert_equal 0 [r XLEN mystream] + assert_equal {0 {} {} {}} [r XPENDING mystream group1] + assert_equal {0 {} {} {}} [r XPENDING mystream group2] + } + + test "XACKDEL with KEEPREF" { + r DEL mystream + r XADD mystream 1-0 f v + r XADD mystream 2-0 f v + + # Create two consumer groups + r XGROUP CREATE mystream group1 0 + r XGROUP CREATE mystream group2 0 + r XREADGROUP GROUP group1 consumer1 STREAMS mystream > + r XREADGROUP GROUP group2 consumer2 STREAMS mystream > + + # Test XACKDEL with KEEPREF + # XACKDEL only deletes the message from the stream + # but does not clean up references in consumer groups' PELs + assert_equal {1 1} [r XACKDEL mystream group1 KEEPREF IDS 2 1-0 2-0] + assert_equal 0 [r XLEN mystream] + assert_equal {0 {} {} {}} [r XPENDING mystream group1] + assert_equal {2 1-0 2-0 {{consumer2 2}}} [r XPENDING mystream group2] + + # Acknowledge remaining messages in group2 + assert_equal {1 1} [r XACKDEL mystream group2 KEEPREF IDS 2 1-0 2-0] + assert_equal {0 {} {} {}} [r XPENDING mystream group1] + assert_equal {0 {} {} {}} [r XPENDING mystream group2] + } + } } diff --git a/tests/unit/type/stream.tcl b/tests/unit/type/stream.tcl index b54b48d15..cbd0f7ce0 100644 --- a/tests/unit/type/stream.tcl +++ b/tests/unit/type/stream.tcl @@ -205,6 +205,64 @@ start_server { } } + test {XADD with MAXLEN option and ACKED option} { + r DEL mystream + r XADD mystream 1-0 f v + r XADD mystream 2-0 f v + r XADD mystream 3-0 f v + r XADD mystream 4-0 f v + r XADD mystream 5-0 f v + assert {[r XLEN mystream] == 5} + + # Create a consumer group but don't read any messages yet + # ACKED option should preserve all messages since none are acked. + r XGROUP CREATE mystream mygroup 0 + r XADD mystream MAXLEN = 1 ACKED 6-0 f v + assert {[r XLEN mystream] == 6} ;# All messages preserved + the new one + + # Read 1 messages and acknowledge them + # This leaves 5 messages still unacked + set records [r XREADGROUP GROUP mygroup consumer1 COUNT 1 STREAMS mystream >] + r XACK mystream mygroup [lindex [lindex [lindex [lindex $records 0] 1] 0] 0] + assert {[lindex [r XPENDING mystream mygroup] 0] == 0} + + # With 5 messages still unacked, ACKED option should preserve them + r XADD mystream MAXLEN = 1 ACKED 7-0 f v + assert {[r XLEN mystream] == 6} ;# 6 - 1 acked + 1 new + + # Acknowledge all remaining messages + set records [r XREADGROUP GROUP mygroup consumer1 STREAMS mystream >] + set ids {} + foreach entry [lindex [lindex $records 0] 1] { + lappend ids [lindex $entry 0] + } + r XACK mystream mygroup {*}$ids + assert {[lindex [r XPENDING mystream mygroup] 0] == 0} ;# All messages acked + + # Now ACKED should trim to MAXLEN since all messages are acked + r XADD mystream MAXLEN = 1 ACKED * f v + assert {[r XLEN mystream] == 1} ;# Successfully trimmed to 1 entries + } + + test {XADD with MAXLEN option and DELREF option} { + r DEL mystream + r XADD mystream 1-0 f v + r XADD mystream 2-0 f v + r XADD mystream 3-0 f v + r XADD mystream 4-0 f v + r XADD mystream 5-0 f v + + r XGROUP CREATE mystream mygroup 0 + r XREADGROUP GROUP mygroup consumer1 COUNT 1 STREAMS mystream > + + # XADD with MAXLEN and DELREF should trim and remove all references + r XADD mystream MAXLEN = 1 DELREF * f v + assert {[r XLEN mystream] == 1} + + # All PEL entries should be cleaned up + assert {[lindex [r XPENDING mystream mygroup] 0] == 0} + } + test {XTRIM with MINID option} { r DEL mystream r XADD mystream 1-0 f v @@ -1081,3 +1139,140 @@ start_server {tags {"stream"}} { assert_match "*wrong number of arguments for 'xinfo|help' command" $e } } + +start_server {tags {"stream"}} { + test "XDELEX wrong number of args" { + assert_error {*wrong number of arguments for 'xdelex' command} {r XDELEX s DELREF} + } + + test "XDELEX should return empty array when key doesn't exist" { + r DEL nonexist + assert_equal {-1 -1} [r XDELEX nonexist IDS 2 1-1 2-2] + } + + test "XDELEX IDS parameter validation" { + r DEL s + r XADD s 1-0 f v + r XGROUP CREATE s g 0 + + # Test invalid numids + assert_error {*Number of IDs must be a positive integer*} {r XDELEX s IDS abc 1-1} + assert_error {*Number of IDs must be a positive integer*} {r XDELEX s IDS 0 1-1} + assert_error {*Number of IDs must be a positive integer*} {r XDELEX s IDS -5 1-1} + + # Test whether numids is equal to the number of IDs provided + assert_error {*The `numids` parameter must match the number of arguments*} {r XDELEX s IDS 3 1-1 2-2} + assert_error {*syntax error*} {r XDELEX s IDS 1 1-1 2-2} + + # Delete non-existent ids + assert_equal {-1 -1} [r XDELEX s IDS 2 1-1 2-2] + } + + test "XDELEX KEEPREF/DELREF/ACKED parameter validation" { + # Test mutually exclusive options + assert_error {*syntax error*} {r XDELEX s KEEPREF DELREF IDS 1 1-1} + assert_error {*syntax error*} {r XDELEX s KEEPREF ACKED IDS 1 1-1} + assert_error {*syntax error*} {r XDELEX s ACKED DELREF IDS 1 1-1} + } + + test "XDELEX with DELREF option acknowledges will remove entry from all PELs" { + r DEL mystream + r XADD mystream 1-0 f v + r XADD mystream 2-0 f v + + # Create two consumer groups + r XGROUP CREATE mystream group1 0 + r XGROUP CREATE mystream group2 0 + r XREADGROUP GROUP group1 consumer1 STREAMS mystream > + r XREADGROUP GROUP group2 consumer2 STREAMS mystream > + + # Verify the message was removed from both groups' PELs when with DELREF + assert_equal {1 1} [r XDELEX mystream DELREF IDS 2 1-0 2-0] + assert_equal 0 [r XLEN mystream] + assert_equal {0 {} {} {}} [r XPENDING mystream group1] + assert_equal {0 {} {} {}} [r XPENDING mystream group2] + } + + test "XDELEX with ACKED option only deletes messages acknowledged by all groups" { + r DEL mystream + r XADD mystream 1-0 f v + r XADD mystream 2-0 f v + + # Create two consumer groups + r XGROUP CREATE mystream group1 0 + r XGROUP CREATE mystream group2 0 + r XGROUP CREATE mystream group3 0 + r XREADGROUP GROUP group1 consumer1 STREAMS mystream > + r XREADGROUP GROUP group2 consumer2 STREAMS mystream > + + # The message is referenced by three consumer groups: + # - group1 and group2 have read the messages + # - group3 hasn't read the messages yet (not delivered) + # Even after group1 acknowledges the messages, they still can't be deleted + r XACK mystream group1 1-0 2-0 + assert_equal {2 2} [r XDELEX mystream ACKED IDS 2 1-0 2-0] + assert_equal 2 [r XLEN mystream] + + # Even after both group1 and group2 acknowledge the messages, these entries + # still can't be deleted because group3 hasn't even read them yet. + r XACK mystream group2 1-0 2-0 + assert_equal {2 2} [r XDELEX mystream ACKED IDS 2 1-0 2-0] + assert_equal 2 [r XLEN mystream] + + # Now group3 reads the messages, but hasn't acknowledged them yet. + # these entries still can't be deleted because group3 hasn't acknowledged them. + r XREADGROUP GROUP group3 consumer3 STREAMS mystream > + assert_equal {2 2} [r XDELEX mystream ACKED IDS 2 1-0 2-0] + assert_equal 2 [r XLEN mystream] + + # Now group3 acknowledges the messages. These entries can now be deleted. + r XACK mystream group3 1-0 2-0 + r XDELEX mystream ACKED IDS 2 1-0 2-0 + assert_equal 0 [r XLEN mystream] + } + + test "XDELEX with ACKED option won't delete messages when new consumer groups are created" { + r DEL mystream + r XADD mystream 1-0 f v + r XADD mystream 2-0 f v + r XADD mystream 3-0 f v + + r XGROUP CREATE mystream group1 0 + r XREADGROUP GROUP group1 consumer1 STREAMS mystream > + + # When the group1 ack message, the message can be deleted with ACK option. + assert_equal {3} [r XACK mystream group1 1-0 2-0 3-0] + assert_equal {1} [r XDELEX mystream ACKED IDS 1 1-0] + + # Create a new consumer group that hasn't read the messages yet. + # Even if group1 ack the message, we still can't delete the message. + r XGROUP CREATE mystream group2 0 + assert_equal {2} [r XDELEX mystream ACKED IDS 1 2-0] + + # Now group2 reads and acknowledges the messages, + # so we can be successfully deleted with the ACKED option. + r XREADGROUP GROUP group2 consumer1 STREAMS mystream > + assert_equal {2} [r XACK mystream group2 2-0 3-0] + assert_equal {1 1} [r XDELEX mystream ACKED IDS 2 2-0 3-0] + } + + test "XDELEX with KEEPREF" { + r DEL mystream + r XADD mystream 1-0 f v + r XADD mystream 2-0 f v + + # Create two consumer groups + r XGROUP CREATE mystream group1 0 + r XGROUP CREATE mystream group2 0 + r XREADGROUP GROUP group1 consumer1 STREAMS mystream > + r XREADGROUP GROUP group2 consumer2 STREAMS mystream > + + # Test XDELEX with KEEPREF + # XDELEX only deletes the message from the stream + # but does not clean up references in consumer groups' PELs + assert_equal {1 1} [r XDELEX mystream KEEPREF IDS 2 1-0 2-0] + assert_equal 0 [r XLEN mystream] + assert_equal {2 1-0 2-0 {{consumer1 2}}} [r XPENDING mystream group1] + assert_equal {2 1-0 2-0 {{consumer2 2}}} [r XPENDING mystream group2] + } +}