diff --git a/src/commands.def b/src/commands.def index fa7cfe4e95c..54de933bd3e 100644 --- a/src/commands.def +++ b/src/commands.def @@ -10452,6 +10452,42 @@ 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,KSPEC_BS_INDEX,.bs.index={1},KSPEC_FK_RANGE,.fk.range={0,1,0}} +}; +#endif + +/* XACKDEL mode argument table */ +struct COMMAND_ARG XACKDEL_mode_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 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("mode",ARG_TYPE_ONEOF,-1,NULL,NULL,NULL,CMD_ARG_OPTIONAL,3,NULL),.subargs=XACKDEL_mode_Subargs}, +{MAKE_ARG("ids",ARG_TYPE_PURE_TOKEN,-1,"IDS",NULL,NULL,CMD_ARG_NONE,0,NULL)}, +{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)}, +}; + /********** XADD ********************/ #ifndef SKIP_CMD_HISTORY_TABLE @@ -10612,6 +10648,41 @@ 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_UPDATE,KSPEC_BS_INDEX,.bs.index={1},KSPEC_FK_RANGE,.fk.range={0,1,0}} +}; +#endif + +/* XDELEX mode argument table */ +struct COMMAND_ARG XDELEX_mode_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 argument table */ +struct COMMAND_ARG XDELEX_Args[] = { +{MAKE_ARG("key",ARG_TYPE_KEY,0,NULL,NULL,NULL,CMD_ARG_NONE,0,NULL)}, +{MAKE_ARG("mode",ARG_TYPE_ONEOF,-1,NULL,NULL,NULL,CMD_ARG_OPTIONAL,3,NULL),.subargs=XDELEX_mode_Subargs}, +{MAKE_ARG("ids",ARG_TYPE_PURE_TOKEN,-1,"IDS",NULL,NULL,CMD_ARG_NONE,0,NULL)}, +{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)}, +}; + /********** XGROUP CREATE ********************/ #ifndef SKIP_CMD_HISTORY_TABLE @@ -12214,10 +12285,12 @@ struct COMMAND_STRUCT serverCommandTable[] = { {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_SLOW|ACL_CATEGORY_SORTEDSET|ACL_CATEGORY_WRITE,NULL,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_FAST|ACL_CATEGORY_STREAM|ACL_CATEGORY_WRITE,NULL,XACK_Keyspecs,1,NULL,3),.args=XACK_Args}, +{MAKE_CMD("xackdel","Acknowledge and (if possible) delete stream message(s).","O(1) for each single item to delete in the stream, regardless of the stream size","9.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_FAST|ACL_CATEGORY_WRITE|ACL_CATEGORY_STREAM,NULL,XACKDEL_Keyspecs,1,NULL,6),.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,2,XADD_Tips,1,xaddCommand,-5,CMD_WRITE|CMD_DENYOOM|CMD_FAST,ACL_CATEGORY_FAST|ACL_CATEGORY_STREAM|ACL_CATEGORY_WRITE,NULL,XADD_Keyspecs,1,NULL,5),.args=XADD_Args}, {MAKE_CMD("xautoclaim","Changes, or acquires, ownership of messages in a consumer group, as if the messages were delivered to a 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_FAST|ACL_CATEGORY_STREAM|ACL_CATEGORY_WRITE,NULL,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 to 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_FAST|ACL_CATEGORY_STREAM|ACL_CATEGORY_WRITE,NULL,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_FAST|ACL_CATEGORY_STREAM|ACL_CATEGORY_WRITE,NULL,XDEL_Keyspecs,1,NULL,2),.args=XDEL_Args}, +{MAKE_CMD("xdelex","Delete stream message(s) with extended options","O(1) for each single item to delete in the stream, regardless of the stream size","9.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_FAST|ACL_CATEGORY_WRITE|ACL_CATEGORY_STREAM,NULL,XDELEX_Keyspecs,1,NULL,5),.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,ACL_CATEGORY_SLOW,NULL,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,ACL_CATEGORY_SLOW,NULL,XINFO_Keyspecs,0,NULL,0),.subcommands=XINFO_Subcommands}, {MAKE_CMD("xlen","Returns 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_FAST|ACL_CATEGORY_READ|ACL_CATEGORY_STREAM,NULL,XLEN_Keyspecs,1,NULL,1),.args=XLEN_Args}, diff --git a/src/commands/xackdel.json b/src/commands/xackdel.json new file mode 100644 index 00000000000..62d0e7ab3e9 --- /dev/null +++ b/src/commands/xackdel.json @@ -0,0 +1,97 @@ +{ + "XACKDEL": { + "summary": "Acknowledge and (if possible) delete stream message(s).", + "complexity": "O(1) for each single item to delete in the stream, regardless of the stream size", + "group": "stream", + "since": "9.2.0", + "arity": -6, + "function": "xackdelCommand", + "command_flags": [ + "WRITE", + "FAST" + ], + "acl_categories": [ + "FAST", + "WRITE", + "STREAM" + ], + "key_specs": [ + { + "flags": [ + "RW", + "UPDATE" + ], + "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": "mode", + "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", + "type": "pure-token", + "token": "IDS" + }, + { + "name": "numids", + "type": "integer" + }, + { + "name": "id", + "type": "string", + "multiple": true + } + ], + "reply_schema": { + "description": "The command returns an integer for each stream message: -1=message not found, 1=acked and deleted, 2=acked but not deleted.", + "type": "array", + "minItems": 1, + "items": { + "description": "Status of the stream message, -1=message not found, 1=acked and deleted, 2=acked but not deleted.", + "type": "integer", + "minimum": -1, + "maximum": 2 + } + } + } +} diff --git a/src/commands/xdelex.json b/src/commands/xdelex.json new file mode 100644 index 00000000000..11fa5a18ae8 --- /dev/null +++ b/src/commands/xdelex.json @@ -0,0 +1,93 @@ +{ + "XDELEX": { + "summary": "Delete stream message(s) with extended options", + "complexity": "O(1) for each single item to delete in the stream, regardless of the stream size", + "group": "stream", + "since": "9.2.0", + "arity": -5, + "function": "xdelexCommand", + "command_flags": [ + "WRITE", + "FAST" + ], + "acl_categories": [ + "FAST", + "WRITE", + "STREAM" + ], + "key_specs": [ + { + "flags": [ + "RW", + "UPDATE" + ], + "begin_search": { + "index": { + "pos": 1 + } + }, + "find_keys": { + "range": { + "lastkey": 0, + "step": 1, + "limit": 0 + } + } + } + ], + "arguments": [ + { + "name": "key", + "type": "key", + "key_spec_index": 0 + }, + { + "name": "mode", + "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": "pure-token" + }, + { + "name": "numids", + "type": "integer" + }, + { + "name": "id", + "type": "string", + "multiple": true + } + ], + "reply_schema": { + "description": "The command returns an integer for each stream message: -1=message not found, 1=message was deleted, 2=message was not deleted due to existing references (ACKED mode).", + "type": "array", + "minItems": 0, + "items": { + "description": "Status of the stream message, -1=message not found, 1=message was deleted, 2=message was not deleted due to existing references (ACKED mode).", + "type": "integer", + "minimum": -1, + "maximum": 2 + } + } + } +} diff --git a/src/server.c b/src/server.c index 0b570208de0..0b71a99dce6 100644 --- a/src/server.c +++ b/src/server.c @@ -2301,6 +2301,8 @@ void createSharedObjects(void) { shared.srem = createSharedString("SREM"); shared.xgroup = createSharedString("XGROUP"); shared.xclaim = createSharedString("XCLAIM"); + shared.xdel = createSharedString("XDEL"); + shared.xack = createSharedString("XACK"); shared.script = createSharedString("SCRIPT"); shared.replconf = createSharedString("REPLCONF"); shared.pexpireat = createSharedString("PEXPIREAT"); diff --git a/src/server.h b/src/server.h index 39e9936add9..dfe7ee854f7 100644 --- a/src/server.h +++ b/src/server.h @@ -1529,7 +1529,7 @@ struct sharedObjectsStruct { *execaborterr, *noautherr, *noreplicaserr, *busykeyerr, *oomerr, *plus, *messagebulk, *pmessagebulk, *subscribebulk, *unsubscribebulk, *psubscribebulk, *punsubscribebulk, *del, *unlink, *rpop, *lpop, *lpush, *zadd, *rpoplpush, *lmove, *blmove, *zpopmin, *zpopmax, *emptyscan, *multi, *exec, *left, *right, *hset, *hsetex, *hdel, *hpexpireat, *hpersist, *srem, - *xgroup, *xclaim, *script, *replconf, *eval, *cluster, *syncslots, *persist, *set, *pexpireat, *pexpire, *time, *pxat, *absttl, + *xgroup, *xclaim, *xdel, *xack, *script, *replconf, *eval, *cluster, *syncslots, *persist, *set, *pexpireat, *pexpire, *time, *pxat, *absttl, *retrycount, *force, *justid, *entriesread, *lastid, *ping, *setid, *keepttl, *load, *createconsumer, *getack, *special_asterisk, *special_equals, *default_username, *redacted, *ssubscribebulk, *sunsubscribebulk, *fields, *finish, *state, *success, *failed, *name, *message, @@ -4345,6 +4345,8 @@ void xclaimCommand(client *c); void xautoclaimCommand(client *c); void xinfoCommand(client *c); void xdelCommand(client *c); +void xackdelCommand(client *c); +void xdelexCommand(client *c); void xtrimCommand(client *c); void lolwutCommand(client *c); void aclCommand(client *c); diff --git a/src/t_stream.c b/src/t_stream.c index af92e6a0316..b9f53998bf3 100644 --- a/src/t_stream.c +++ b/src/t_stream.c @@ -1625,6 +1625,47 @@ void streamPropagateConsumerCreation(client *c, robj *key, robj *groupname, sds decrRefCount(argv[4]); } +/* Propagate the deletion of stream entries as + * + * XDEL ... + * + * XDELEX & XACKDEL propagate their effects manually this way to ensure + * compatibility with any pre-9.2 replicas. */ +static void streamPropagateDelIDs(client *c, robj *key, streamID *ids, int count) { + if (count == 0) return; + + robj **argv = zmalloc(sizeof(robj *) * (2 + count)); + argv[0] = shared.xdel; + argv[1] = key; + for (int j = 0; j < count; j++) argv[2 + j] = createObjectFromStreamID(&ids[j]); + + alsoPropagate(c->db->id, argv, 2 + count, PROPAGATE_AOF | PROPAGATE_REPL, c->slot); + + for (int j = 0; j < count; j++) decrRefCount(argv[2 + j]); + zfree(argv); +} + +/* Propagate acknowledgement of 'count' ids for 'groupname' as + * + * XACK ... + * + * XDELEX & XACKDEL propagate their effects manually this way to ensure + * compatibility with any pre-9.2 replicas. */ +static void streamPropagateAckIDs(client *c, robj *key, robj *groupname, streamID *ids, int count) { + if (count == 0) return; + + robj **argv = zmalloc(sizeof(robj *) * (3 + count)); + argv[0] = shared.xack; + argv[1] = key; + argv[2] = groupname; + for (int j = 0; j < count; j++) argv[3 + j] = createObjectFromStreamID(&ids[j]); + + alsoPropagate(c->db->id, argv, 3 + count, PROPAGATE_AOF | PROPAGATE_REPL, c->slot); + + for (int j = 0; j < count; j++) decrRefCount(argv[3 + j]); + zfree(argv); +} + /* Send the stream items in the specified range to the client 'c'. The range * the client will receive is between start and end inclusive, if 'count' is * non zero, no more than 'count' elements are sent. @@ -2520,6 +2561,21 @@ void streamFreeNACK(streamNACK *na) { zfree(na); } +/* Delete a pending entry from the group PEL and from the PEL of the consumer + * owning it, freeing the NACK. Returns 1 if entry was pending and was deleted, + * 0 otherwise leaving both group & individual consumer PEL untouched. */ +static int streamDeletePELEntry(rax *pel, streamID *id) { + unsigned char buf[sizeof(streamID)]; + streamEncodeID(buf, id); + void *result; + if (!raxFind(pel, buf, sizeof(buf), &result)) return 0; + streamNACK *nack = result; + raxRemove(pel, buf, sizeof(buf), NULL); + raxRemove(nack->consumer->pel, buf, sizeof(buf), NULL); + streamFreeNACK(nack); + return 1; +} + /* Free a consumer and associated data structures. Note that this function * will not reassign the pending messages associated with this consumer * nor will delete them from the stream, so when this function is called @@ -2904,12 +2960,7 @@ void xackCommand(client *c) { /* 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); - streamFreeNACK(nack); + if (streamDeletePELEntry(group->pel, &ids[j - 3])) { acknowledged++; server.dirty++; } @@ -3551,48 +3602,294 @@ void xautoclaimCommand(client *c) { preventCommandPropagation(c); } +/* PEL handling modes shared by XDELEX & XACKDEL. */ +typedef enum { + PELMODE_KEEPREF = 0, + PELMODE_DELREF, + PELMODE_ACKED +} streamPELMode; + +/* Command variant for xdelGenericCommand. */ +typedef enum { + XDEL_CMD, /* XDEL ... */ + XDELEX_CMD, /* XDELEX [KEEPREF|DELREF|ACKED] IDS ... */ + XACKDEL_CMD, /* XACKDEL [KEEPREF|DELREF|ACKED] IDS ... */ +} xdelVariant; + /* XDEL [ ... ] + * XDELEX [KEEPREF | DELREF | ACKED] IDS num [ ... ] + * XACKDEL [KEEPREF | DELREF | ACKED] IDS num [ ... ] * - * Removes the specified entries from the stream. Returns the number - * of items actually deleted, that may be different from the number - * of IDs passed in case certain IDs do not exist. */ -void xdelCommand(client *c) { - robj *o; + * Unified implementation of XDEL, XDELEX and XACKDEL. + * + * XDEL removes stream entries unconditionally. + * XDELEX is XDEL with PEL-awareness across all consumer groups. + * XACKDEL is XDELEX scoped to a target consumer group: it acknowledges + * entries in the target group first, then consults remaining groups. */ +static void xdelGenericCommand(client *c, xdelVariant variant) { + bool has_group = (variant == XACKDEL_CMD); + bool has_pelmode = (variant != XDEL_CMD); + bool array_reply = (variant != XDEL_CMD); + + /* --- Argument parsing ------------------------------------------------ */ + streamCG *group = NULL; + streamPELMode mode = PELMODE_KEEPREF; + robj *o = lookupKeyWrite(c->db, c->argv[1]); + int argi = 2; - if ((o = lookupKeyWriteOrReply(c, c->argv[1], shared.czero)) == NULL || checkType(c, o, OBJ_STREAM)) return; + if (o && checkType(c, o, OBJ_STREAM)) return; /* Type error. */ + + if (has_group) { + /* The group name is a positional argument: always consume it, even + * when the key is missing (the lookup simply yields a NULL group). */ + if (o) { + group = streamLookupCG(objectGetVal(o), objectGetVal(c->argv[argi])); + } + argi++; /* past group */ + } + + size_t id_count; + if (!has_pelmode) { + /* XDEL has no IDS token, so the remaining args is the id count. */ + id_count = c->argc - argi; + } else { + /* Parse optional PEL mode: [KEEPREF | DELREF | ACKED] */ + if (strcasecmp(objectGetVal(c->argv[argi]), "KEEPREF") == 0) { + argi++; + } else if (strcasecmp(objectGetVal(c->argv[argi]), "DELREF") == 0) { + argi++; + mode = PELMODE_DELREF; + } else if (strcasecmp(objectGetVal(c->argv[argi]), "ACKED") == 0) { + argi++; + mode = PELMODE_ACKED; + } + + /* Expect IDS token. */ + if (strcasecmp(objectGetVal(c->argv[argi]), "IDS") != 0) { + addReplyErrorObject(c, shared.syntaxerr); + return; + } + argi++; /* past IDS */ + + /* Parse and validate numids: must be a positive integer. */ + long long ll; + if (getLongLongFromObject(c->argv[argi], &ll) != C_OK || ll <= 0) { + addReplyError(c, "Number of IDs must be a positive integer"); + return; + } + argi++; /* past numids */ + + /* Validate numids matches remaining arg count. */ + if (ll != c->argc - argi) { + addReplyErrorObject(c, shared.syntaxerr); + return; + } + id_count = (size_t)ll; + } + + /* --- Missing key / group early exit ---------------------------------- */ + if (o == NULL || (has_group && group == NULL)) { + if (array_reply) { + addReplyArrayLen(c, id_count); + for (size_t i = 0; i < id_count; i++) addReplyLongLong(c, -1); + } else { + addReply(c, shared.czero); + } + return; + } stream *s = objectGetVal(o); + /* --- Allocate working arrays ----------------------------------------- * + * Each variant only declares static buffers for the arrays it actually + * uses. Unused pointers are NULL so accidental access crashes rather + * than silently touching an unrelated stack buffer. For large id_count + * the heap path also skips allocations the variant does not need. */ + streamID static_ids[STREAMID_STATIC_VECTOR_LEN]; + streamID *ids = static_ids; + + streamID static_del_ids[STREAMID_STATIC_VECTOR_LEN]; + streamID *del_ids = static_del_ids; + int del_count = 0; + + int static_resps[STREAMID_STATIC_VECTOR_LEN]; + int *resps = array_reply ? static_resps : NULL; + + unsigned char static_acked_flags[STREAMID_STATIC_VECTOR_LEN]; + unsigned char *acked_flags = has_group ? static_acked_flags : NULL; + + unsigned char static_exists[STREAMID_STATIC_VECTOR_LEN]; + unsigned char *exists = (mode == PELMODE_ACKED) ? static_exists : NULL; + + unsigned char static_cleared[STREAMID_STATIC_VECTOR_LEN]; + unsigned char *cleared = (mode == PELMODE_DELREF || mode == PELMODE_ACKED) ? static_cleared : NULL; + + streamID static_ack_ids[STREAMID_STATIC_VECTOR_LEN]; + streamID *ack_ids = (has_group || mode == PELMODE_DELREF || mode == PELMODE_ACKED) ? static_ack_ids : NULL; + + if (id_count > STREAMID_STATIC_VECTOR_LEN) { + ids = zmalloc(sizeof(streamID) * id_count); + del_ids = zmalloc(sizeof(streamID) * id_count); + if (resps) resps = zmalloc(sizeof(int) * id_count); + if (acked_flags) acked_flags = zmalloc(sizeof(unsigned char) * id_count); + if (exists) exists = zmalloc(sizeof(unsigned char) * id_count); + if (cleared) cleared = zmalloc(sizeof(unsigned char) * id_count); + if (ack_ids) ack_ids = zmalloc(sizeof(streamID) * id_count); + } + /* 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; - int id_count = c->argc - 2; - if (id_count > STREAMID_STATIC_VECTOR_LEN) ids = zmalloc(sizeof(streamID) * id_count); - for (int j = 2; j < c->argc; j++) { - if (streamParseStrictIDOrReply(c, c->argv[j], &ids[j - 2], 0, NULL) != C_OK) goto cleanup; + for (size_t j = 0; j < id_count; j++) { + if (streamParseStrictIDOrReply(c, c->argv[argi + j], &ids[j], 0, NULL) != C_OK) goto cleanup; + if (array_reply) resps[j] = 1; } - /* Actually apply the command. */ + int acked = 0; int deleted = 0; - int first_entry = 0; - for (int j = 2; j < c->argc; j++) { - streamID *id = &ids[j - 2]; - 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; + bool first_entry = 0; + if (acked_flags) memset(acked_flags, 0, id_count); + + /* --- KEEPREF fast path ----------------------------------------------- * + * When the mode is KEEPREF: with a target group we gate deletion on the + * entry being pending in that group's PEL; without a group we delete + * unconditionally (the original XDEL / XDELEX KEEPREF behaviour). */ + if (mode == PELMODE_KEEPREF) { + for (size_t j = 0; j < id_count; j++) { + streamID *id = &ids[j]; + + if (group) { + /* XACKDEL KEEPREF: only delete if pending in target group. */ + if (!streamDeletePELEntry(group->pel, id)) { + if (array_reply) resps[j] = -1; + continue; + } + acked++; + acked_flags[j] = 1; } - /* Update the stream's maximal tombstone if needed. */ - if (streamCompareID(id, &s->max_deleted_entry_id) > 0) { - s->max_deleted_entry_id = *id; + + if (streamDeleteItem(s, id)) { + deleted++; + del_ids[del_count++] = *id; + if (streamCompareID(id, &s->first_id) == 0) first_entry = 1; + if (streamCompareID(id, &s->max_deleted_entry_id) > 0) s->max_deleted_entry_id = *id; + } else if (array_reply) { + /* Entry does not exist in the stream. */ + resps[j] = -1; } - deleted++; - }; + } + goto sync; + } + + /* --- Phase 1 (XACKDEL only): target-group PEL scan ------------------- * + * If the entry isn't pending in the target group we mark it -1 and skip + * it in Phase 2. XDELEX has no target group so this phase is skipped. */ + if (group) { + for (size_t j = 0; j < id_count; j++) { + if (streamDeletePELEntry(group->pel, &ids[j])) { + acked++; + acked_flags[j] = 1; + /* resps[j] stays 1: eligible for deletion. */ + } else { + resps[j] = -1; + } + } + } + + /* --- Phase 2: iterate consumer groups -------------------------------- * + * XDELEX iterates all groups uniformly; XACKDEL skips the target group + * (handled in Phase 1). The `if (cg == group) continue` naturally + * never fires when group is NULL (XDELEX). */ + if ((mode == PELMODE_DELREF || mode == PELMODE_ACKED) && s->cgroups != NULL) { + memset(cleared, 0, id_count); + + /* Determine stream message existence upfront for ACKED mode. */ + if (mode == PELMODE_ACKED) { + for (size_t j = 0; j < id_count; j++) { + exists[j] = streamEntryExists(s, &ids[j]); + } + } + + raxIterator ri_cgroups; + raxStart(&ri_cgroups, s->cgroups); + raxSeek(&ri_cgroups, "^", NULL, 0); + while (raxNext(&ri_cgroups)) { + streamCG *cg = ri_cgroups.data; + if (cg == group) continue; /* Target group handled in Phase 1. */ + + for (size_t j = 0; j < id_count; j++) { + if (resps[j] != 1) continue; /* Already finalized. */ + if (mode == PELMODE_ACKED && resps[j] == 2) continue; + + streamID *id = &ids[j]; + + if (mode == PELMODE_DELREF) { + if (streamDeletePELEntry(cg->pel, id)) { + acked++; + cleared[j] = 1; + } + } else { + /* ACKED: check PEL before consulting cg->last_id. */ + unsigned char buf[sizeof(streamID)]; + streamEncodeID(buf, id); + void *result; + if (raxFind(cg->pel, buf, sizeof(buf), &result)) { + resps[j] = 2; + } else if (exists[j] && + streamCompareID(id, &cg->last_id) > 0) { + resps[j] = 2; + } + } + } + + if (mode == PELMODE_DELREF) { + int ack_count = 0; + for (size_t j = 0; j < id_count; j++) { + if (cleared[j]) { + ack_ids[ack_count++] = ids[j]; + cleared[j] = 0; + } + } + if (ack_count) { + robj *groupname = createStringObject((char *)ri_cgroups.key, ri_cgroups.key_len); + streamPropagateAckIDs(c, c->argv[1], groupname, ack_ids, ack_count); + decrRefCount(groupname); + } + } + } + raxStop(&ri_cgroups); + + /* ACKED without a target group: entries that don't exist and that no + * group references return "not found". When a target group exists + * Phase 1 already marked non-pending entries as -1. */ + if (mode == PELMODE_ACKED && !group) { + for (size_t j = 0; j < id_count; j++) { + if (resps[j] == 1 && !exists[j]) resps[j] = -1; + } + } } - /* Update the stream's first ID. */ + /* --- Deletion phase -------------------------------------------------- * + * Delete entries whose status is still 1 (eligible). */ + for (size_t j = 0; j < id_count; j++) { + if (resps[j] == 1) { + streamID *id = &ids[j]; + if (streamDeleteItem(s, id)) { + deleted++; + del_ids[del_count++] = *id; + if (streamCompareID(id, &s->first_id) == 0) first_entry = 1; + if (streamCompareID(id, &s->max_deleted_entry_id) > 0) s->max_deleted_entry_id = *id; + } else if (!acked_flags || !acked_flags[j]) { + /* Entry doesn't exist and was never pending in the target + * group — genuinely not found. When acked_flags[j] is set + * the target-group PEL was successfully cleared in Phase 1, + * so the entry being already gone is fine (status stays 1). */ + resps[j] = -1; + } + } + } + +sync: + /* --- Stream bookkeeping & signalling --------------------------------- */ if (deleted) { if (s->length == 0) { s->first_id.ms = 0; @@ -3601,16 +3898,64 @@ void xdelCommand(client *c) { streamGetEdgeID(s, 1, 1, &s->first_id); } } - - /* Propagate the write if needed. */ - if (deleted) { + if (deleted || acked) { signalModifiedKey(c, c->db, c->argv[1]); notifyKeyspaceEvent(NOTIFY_STREAM, "xdel", c->argv[1], c->db->id); server.dirty += deleted; } - addReplyLongLong(c, deleted); + if (acked) { + server.dirty += acked; + } + + /* --- Propagation ----------------------------------------------------- * + * XDELEX/XACKDEL are rewritten as XACK + XDEL primitives so that + * pre-9.2 replicas can apply them. XDEL propagates as itself. */ + if (has_pelmode) { + preventCommandPropagation(c); + + /* Target-group acknowledgements (XACKDEL only). */ + if (group) { + int ack_count = 0; + for (size_t j = 0; j < id_count; j++) { + if (acked_flags[j]) ack_ids[ack_count++] = ids[j]; + } + streamPropagateAckIDs(c, c->argv[1], c->argv[2], ack_ids, ack_count); + } + + streamPropagateDelIDs(c, c->argv[1], del_ids, del_count); + } + + /* --- Reply ----------------------------------------------------------- */ + if (array_reply) { + addReplyArrayLen(c, id_count); + for (size_t j = 0; j < id_count; j++) addReplyLongLong(c, resps[j]); + } else { + addReplyLongLong(c, deleted); + } + cleanup: if (ids != static_ids) zfree(ids); + if (resps != static_resps) zfree(resps); + if (acked_flags != static_acked_flags) zfree(acked_flags); + if (exists != static_exists) zfree(exists); + if (cleared != static_cleared) zfree(cleared); + if (ack_ids != static_ack_ids) zfree(ack_ids); + if (del_ids != static_del_ids) zfree(del_ids); +} + +/* XDEL [ ... ] */ +void xdelCommand(client *c) { + xdelGenericCommand(c, XDEL_CMD); +} + +/* XDELEX [KEEPREF | DELREF | ACKED] IDS num [ ... ] */ +void xdelexCommand(client *c) { + xdelGenericCommand(c, XDELEX_CMD); +} + +/* XACKDEL [KEEPREF | DELREF | ACKED] IDS num [ ... ] */ +void xackdelCommand(client *c) { + xdelGenericCommand(c, XACKDEL_CMD); } /* General form: XTRIM [... options ...] diff --git a/tests/integration/cross-version-replication.tcl b/tests/integration/cross-version-replication.tcl index 0e3de03de3f..48ee1d89c49 100644 --- a/tests/integration/cross-version-replication.tcl +++ b/tests/integration/cross-version-replication.tcl @@ -87,4 +87,48 @@ start_server {tags {"repl needs:other-server external:skip"}} { assert_equal value1 [$old_replica hget hfe field1] } } + + test "XACKDEL replicates as equivalent pre-9.2 commands XACK/XDEL for backwards compatibility" { + if {[version_greater_or_equal $old_replica_version 9.2.0]} { + skip "Replica $old_replica_version must be before 9.2.0 for this test" + } + + r FLUSHALL + r XADD mystream 1-0 hello world + r XGROUP CREATE mystream grp1 0 + r XGROUP CREATE mystream grp2 0 + r XREADGROUP GROUP grp1 alice COUNT 1 STREAMS mystream > + r XREADGROUP GROUP grp2 bob COUNT 1 STREAMS mystream > + r XACKDEL mystream grp1 DELREF IDS 1 1-0 + start_server {start-other-server 1 config "minimal.conf"} { + set old_replica [srv 0 client] + $old_replica replicaof $primary_host $primary_port + wait_for_sync $old_replica 500 100 + assert_equal [llength [$old_replica XRANGE mystream - +]] 0 + assert_equal [llength [$old_replica XPENDING mystream grp1 - + 10]] 0 + assert_equal [llength [$old_replica XPENDING mystream grp2 - + 10]] 0 + } + } + + test "XDELEX replicates as equivalent pre-9.2 commands XACK/XDEL for backwards compatibility" { + if {[version_greater_or_equal $old_replica_version 9.2.0]} { + skip "Replica $old_replica_version must be before 9.2.0 for this test" + } + + r FLUSHALL + r XADD mystream 1-0 hello world + r XGROUP CREATE mystream grp1 0 + r XGROUP CREATE mystream grp2 0 + r XREADGROUP GROUP grp1 alice COUNT 1 STREAMS mystream > + r XREADGROUP GROUP grp2 bob COUNT 1 STREAMS mystream > + r XDELEX mystream DELREF IDS 1 1-0 + start_server {start-other-server 1 config "minimal.conf"} { + set old_replica [srv 0 client] + $old_replica replicaof $primary_host $primary_port + wait_for_sync $old_replica 500 100 + assert_equal [llength [$old_replica XRANGE mystream - +]] 0 + assert_equal [llength [$old_replica XPENDING mystream grp1 - + 10]] 0 + assert_equal [llength [$old_replica XPENDING mystream grp2 - + 10]] 0 + } + } } diff --git a/tests/unit/type/stream-cgroups.tcl b/tests/unit/type/stream-cgroups.tcl index 047defecf74..0cf4a0a307d 100644 --- a/tests/unit/type/stream-cgroups.tcl +++ b/tests/unit/type/stream-cgroups.tcl @@ -1479,6 +1479,357 @@ start_server { } } + start_server {tags {"external:skip"}} { + set master [srv -1 client] + set master_host [srv -1 host] + set master_port [srv -1 port] + set replica [srv 0 client] + + test {XACKDEL replication: ack-only (no deletion) propagates PEL removal to replica} { + $replica replicaof $master_host $master_port + wait_for_condition 50 100 { + [s 0 master_link_status] eq {up} + } else { + fail "Replication not started." + } + + # Two groups both read the message so grp2 blocks deletion in ACKED mode + $master DEL stream + $master XADD stream 1-0 f v + $master XGROUP CREATE stream grp1 0 + $master XGROUP CREATE stream grp2 0 + $master XREADGROUP GROUP grp1 alice COUNT 1 STREAMS stream > + $master XREADGROUP GROUP grp2 bob COUNT 1 STREAMS stream > + + wait_for_ofs_sync $master $replica + + # Replica should have the pending entry in grp1 before ack + assert_equal [llength [$replica XPENDING stream grp1 - + 10]] 1 + + # ACKED mode: grp2 still has entry in PEL so deletion is suppressed + $master XACKDEL stream grp1 ACKED IDS 1 1-0 + + wait_for_ofs_sync $master $replica + + # grp1's PEL entry should be gone, stream entry still present on replica + assert_equal [llength [$replica XPENDING stream grp1 - + 10]] 0 + assert_equal [llength [$replica XRANGE stream - +]] 1 + } + + test {XACKDEL replication: ACKED mode deletion propagates stream removal to replica} { + $replica replicaof $master_host $master_port + wait_for_condition 50 100 { + [s 0 master_link_status] eq {up} + } else { + fail "Replication not started." + } + + $master DEL stream + $master XADD stream 1-0 f v + $master XGROUP CREATE stream grp 0 + $master XREADGROUP GROUP grp alice COUNT 1 STREAMS stream > + + wait_for_ofs_sync $master $replica + + # Replica should have entry in stream and PEL before ack + assert_equal [llength [$replica XRANGE stream - +]] 1 + assert_equal [llength [$replica XPENDING stream grp - + 10]] 1 + + # ACKED mode with only one group: triggers deletion + $master XACKDEL stream grp ACKED IDS 1 1-0 + + wait_for_ofs_sync $master $replica + + # Both PEL entry and stream entry should be gone on replica + assert_equal [llength [$replica XPENDING stream grp - + 10]] 0 + assert_equal [llength [$replica XRANGE stream - +]] 0 + } + + test {XACKDEL replication: DELREF clears other groups' PELs on replica} { + $replica replicaof $master_host $master_port + wait_for_condition 50 100 { + [s 0 master_link_status] eq {up} + } else { + fail "Replication not started." + } + + $master DEL stream + $master XADD stream 1-0 f v + $master XGROUP CREATE stream grp1 0 + $master XGROUP CREATE stream grp2 0 + $master XREADGROUP GROUP grp1 alice COUNT 1 STREAMS stream > + $master XREADGROUP GROUP grp2 bob COUNT 1 STREAMS stream > + + wait_for_ofs_sync $master $replica + + # Both groups should have the entry in their PEL on replica + assert_equal [llength [$replica XPENDING stream grp1 - + 10]] 1 + assert_equal [llength [$replica XPENDING stream grp2 - + 10]] 1 + + # DELREF: ack for grp1, force-delete from stream and clear all groups' PELs + $master XACKDEL stream grp1 DELREF IDS 1 1-0 + + wait_for_ofs_sync $master $replica + + # Stream entry and grp2's PEL entry should both be gone on replica + assert_equal [llength [$replica XRANGE stream - +]] 0 + assert_equal [llength [$replica XPENDING stream grp2 - + 10]] 0 + } + } + + start_server {tags {"external:skip"}} { + set master [srv -1 client] + set master_host [srv -1 host] + set master_port [srv -1 port] + set replica [srv 0 client] + + test {XDELEX replication: KEEPREF deletes stream entry but keeps dangling PEL ref on replica} { + $replica replicaof $master_host $master_port + wait_for_condition 50 100 { + [s 0 master_link_status] eq {up} + } else { + fail "Replication not started." + } + + $master DEL stream + $master XADD stream 1-0 f v + $master XADD stream 2-0 f v + $master XGROUP CREATE stream grp 0 + $master XREADGROUP GROUP grp alice COUNT 1 STREAMS stream > + + wait_for_ofs_sync $master $replica + + # Replica has both entries and grp's PEL contains 1-0 + assert_equal 2 [$replica XLEN stream] + assert_equal 1 [llength [$replica XPENDING stream grp - + 10]] + + # KEEPREF: deletes entry from stream but leaves PEL reference intact + $master XDELEX stream KEEPREF IDS 1 1-0 + + wait_for_ofs_sync $master $replica + + # Stream entry gone on replica, PEL reference still present + assert_equal 1 [$replica XLEN stream] + assert_equal 1 [llength [$replica XPENDING stream grp - + 10]] + } + + test {XDELEX replication: DELREF deletes stream entry and clears PEL on replica} { + $replica replicaof $master_host $master_port + wait_for_condition 50 100 { + [s 0 master_link_status] eq {up} + } else { + fail "Replication not started." + } + + $master DEL stream + $master XADD stream 1-0 f v + $master XADD stream 2-0 f v + $master XGROUP CREATE stream grp 0 + $master XREADGROUP GROUP grp alice COUNT 1 STREAMS stream > + + wait_for_ofs_sync $master $replica + + # Replica has both entries and grp's PEL contains 1-0 + assert_equal 2 [$replica XLEN stream] + assert_equal 1 [llength [$replica XPENDING stream grp - + 10]] + + # DELREF: deletes entry from stream AND removes it from all PELs + $master XDELEX stream DELREF IDS 1 1-0 + + wait_for_ofs_sync $master $replica + + # Stream entry gone and PEL cleared on replica + assert_equal 1 [$replica XLEN stream] + assert_equal 0 [llength [$replica XPENDING stream grp - + 10]] + } + + test {XDELEX replication: ACKED skips pending entries, deletes only after all groups ack} { + $replica replicaof $master_host $master_port + wait_for_condition 50 100 { + [s 0 master_link_status] eq {up} + } else { + fail "Replication not started." + } + + $master DEL stream + $master XADD stream 1-0 f v + $master XGROUP CREATE stream grp 0 + $master XREADGROUP GROUP grp alice COUNT 1 STREAMS stream > + + wait_for_ofs_sync $master $replica + + # Entry is pending; ACKED mode should not delete it + $master XDELEX stream ACKED IDS 1 1-0 + + wait_for_ofs_sync $master $replica + + # Entry still present on replica since it was not acked + assert_equal 1 [$replica XLEN stream] + assert_equal 1 [llength [$replica XPENDING stream grp - + 10]] + + # Now ack the entry and retry XDELEX ACKED + $master XACK stream grp 1-0 + $master XDELEX stream ACKED IDS 1 1-0 + + wait_for_ofs_sync $master $replica + + # Entry deleted on replica after all groups have acked + assert_equal 0 [$replica XLEN stream] + assert_equal 0 [llength [$replica XPENDING stream grp - + 10]] + } + } + + start_server {tags {"external:skip"}} { + set master [srv -1 client] + set master_host [srv -1 host] + set master_port [srv -1 port] + set replica [srv 0 client] + + # Number of times 'cmd' was executed on the replica, or 0 if never + # called (INFO omits zero counters). + proc get_replica_calls {client cmd} { + set info [$client INFO commandstats] + foreach line [split $info "\n"] { + if {[string match "cmdstat_$cmd:*" $line]} { + regexp {calls=(\d+)} $line -> count + return $count + } + } + return 0 + } + + $replica replicaof $master_host $master_port + wait_for_condition 50 100 { + [s 0 master_link_status] eq {up} + } else { + fail "Replication not started." + } + + test {XACKDEL ack-only propagates XACK but never XACKDEL or XDEL} { + $master DEL stream + $master XADD stream 1-0 f v + $master XGROUP CREATE stream grp1 0 + $master XGROUP CREATE stream grp2 0 + $master XREADGROUP GROUP grp1 alice COUNT 1 STREAMS stream > + $master XREADGROUP GROUP grp2 bob COUNT 1 STREAMS stream > + wait_for_ofs_sync $master $replica + + set xack_before [get_replica_calls $replica xack] + set xdel_before [get_replica_calls $replica xdel] + + # grp2 still holds the message pending, so nothing is deleted + $master XACKDEL stream grp1 ACKED IDS 1 1-0 + wait_for_ofs_sync $master $replica + + assert_equal 1 [expr {[get_replica_calls $replica xack] - $xack_before}] + assert_equal 0 [expr {[get_replica_calls $replica xdel] - $xdel_before}] + assert_equal 0 [get_replica_calls $replica xackdel] + assert_equal 1 [$replica XLEN stream] + assert_equal 0 [llength [$replica XPENDING stream grp1 - + 10]] + } + + test {XACKDEL KEEPREF propagates XACK + XDEL but never XACKDEL} { + $master DEL stream + $master XADD stream 1-0 f v + $master XGROUP CREATE stream grp 0 + $master XREADGROUP GROUP grp alice COUNT 1 STREAMS stream > + wait_for_ofs_sync $master $replica + + set xack_before [get_replica_calls $replica xack] + set xdel_before [get_replica_calls $replica xdel] + + $master XACKDEL stream grp KEEPREF IDS 1 1-0 + wait_for_ofs_sync $master $replica + + assert_equal 1 [expr {[get_replica_calls $replica xack] - $xack_before}] + assert_equal 1 [expr {[get_replica_calls $replica xdel] - $xdel_before}] + assert_equal 0 [get_replica_calls $replica xackdel] + assert_equal 0 [$replica XLEN stream] + assert_equal 0 [llength [$replica XPENDING stream grp - + 10]] + } + + test {XACKDEL DELREF propagates per-group XACK + XDEL but never XACKDEL} { + $master DEL stream + $master XADD stream 1-0 f v + $master XGROUP CREATE stream grp1 0 + $master XGROUP CREATE stream grp2 0 + $master XREADGROUP GROUP grp1 alice COUNT 1 STREAMS stream > + $master XREADGROUP GROUP grp2 bob COUNT 1 STREAMS stream > + wait_for_ofs_sync $master $replica + + set xack_before [get_replica_calls $replica xack] + set xdel_before [get_replica_calls $replica xdel] + + $master XACKDEL stream grp1 DELREF IDS 1 1-0 + wait_for_ofs_sync $master $replica + + # One XACK for the target group + one for grp2's cleared PEL ref + assert_equal 2 [expr {[get_replica_calls $replica xack] - $xack_before}] + assert_equal 1 [expr {[get_replica_calls $replica xdel] - $xdel_before}] + assert_equal 0 [get_replica_calls $replica xackdel] + assert_equal 0 [$replica XLEN stream] + assert_equal 0 [llength [$replica XPENDING stream grp2 - + 10]] + } + + test {XDELEX KEEPREF propagates XDEL only but never XDELEX} { + $master DEL stream + $master XADD stream 1-0 f v + wait_for_ofs_sync $master $replica + + set xack_before [get_replica_calls $replica xack] + set xdel_before [get_replica_calls $replica xdel] + + $master XDELEX stream KEEPREF IDS 1 1-0 + wait_for_ofs_sync $master $replica + + assert_equal 1 [expr {[get_replica_calls $replica xdel] - $xdel_before}] + assert_equal 0 [expr {[get_replica_calls $replica xack] - $xack_before}] + assert_equal 0 [get_replica_calls $replica xdelex] + assert_equal 0 [$replica XLEN stream] + } + + test {XDELEX DELREF propagates XDEL + XACK but never XDELEX} { + $master DEL stream + $master XADD stream 1-0 f v + $master XGROUP CREATE stream grp 0 + $master XREADGROUP GROUP grp alice COUNT 1 STREAMS stream > + wait_for_ofs_sync $master $replica + + set xack_before [get_replica_calls $replica xack] + set xdel_before [get_replica_calls $replica xdel] + + $master XDELEX stream DELREF IDS 1 1-0 + wait_for_ofs_sync $master $replica + + assert_equal 1 [expr {[get_replica_calls $replica xdel] - $xdel_before}] + assert_equal 1 [expr {[get_replica_calls $replica xack] - $xack_before}] + assert_equal 0 [get_replica_calls $replica xdelex] + assert_equal 0 [$replica XLEN stream] + assert_equal 0 [llength [$replica XPENDING stream grp - + 10]] + } + + test {XDELEX ACKED with nothing deleted propagates nothing} { + $master DEL stream + $master XADD stream 1-0 f v + $master XGROUP CREATE stream grp 0 + $master XREADGROUP GROUP grp alice COUNT 1 STREAMS stream > + wait_for_ofs_sync $master $replica + + set xack_before [get_replica_calls $replica xack] + set xdel_before [get_replica_calls $replica xdel] + + # Entry still pending in grp, so ACKED neither deletes nor clears + $master XDELEX stream ACKED IDS 1 1-0 + wait_for_ofs_sync $master $replica + + assert_equal 0 [expr {[get_replica_calls $replica xdel] - $xdel_before}] + assert_equal 0 [expr {[get_replica_calls $replica xack] - $xack_before}] + assert_equal 0 [get_replica_calls $replica xdelex] + assert_equal 1 [$replica XLEN stream] + assert_equal 1 [llength [$replica XPENDING stream grp - + 10]] + } + } + start_server {tags {"stream needs:debug"} overrides {appendonly yes aof-use-rdb-preamble no}} { test {Empty stream with no lastid can be rewrite into AOF correctly} { r XGROUP CREATE mystream group-name $ MKSTREAM diff --git a/tests/unit/type/stream.tcl b/tests/unit/type/stream.tcl index 1c8aa533f2f..4440279b18e 100644 --- a/tests/unit/type/stream.tcl +++ b/tests/unit/type/stream.tcl @@ -634,6 +634,928 @@ start_server { } } + test {XACKDEL returns syntax error when IDS token is missing after mode} { + r DEL teststream + r XADD teststream 1 msg hello + r XGROUP CREATE teststream testgrp 0 + assert_error "*syntax error*" {r XACKDEL teststream testgrp KEEPREF NOIDS 1 1} + assert_error "*syntax error*" {r XACKDEL teststream testgrp DELREF NOIDS 1 1} + assert_error "*syntax error*" {r XACKDEL teststream testgrp ACKED NOIDS 1 1} + } + + test {XACKDEL wrong number of args} { + assert_error {*wrong number of arguments*} {r XACKDEL s} + assert_error {*wrong number of arguments*} {r XACKDEL s grp} + } + + test {XACKDEL w/ KEEPREF keeps refs in other consumer groups' PEL} { + r DEL testxadstream + r XADD testxadstream 1 msg hello + r XGROUP CREATE testxadstream testxadgrp1 1 + r XGROUP CREATE testxadstream testxadgrp2 1 + r XADD testxadstream 2 msg2 hello2 + + # Setup consumer groups w/ message in both groups PEL + r XREADGROUP GROUP testxadgrp1 testxadcnsmr COUNT 1 STREAMS testxadstream > + r XREADGROUP GROUP testxadgrp2 testxadcnsmr COUNT 1 STREAMS testxadstream > + + # Group 1 + set pend [r XPENDING testxadstream testxadgrp1] + assert_equal 2-0 [lindex $pend 1] + set ids [r XACKDEL testxadstream testxadgrp1 KEEPREF IDS 1 2] + assert_equal 1 [llength $ids] + assert_equal 1 [lindex $ids 0] + assert_equal {} [r xrange testxadstream 2 2] + + # Group 2 still has ref in PEL + set pend [r XPENDING testxadstream testxadgrp2] + assert_equal 2-0 [lindex $pend 1] + } + + test {XACKDEL uses KEEPREF by default} { + r DEL testxadstream + r XADD testxadstream 1 msg hello + r XGROUP CREATE testxadstream testxadgrp1 1 + r XGROUP CREATE testxadstream testxadgrp2 1 + r XADD testxadstream 2 msg2 hello2 + + # Setup consumer groups w/ message in both groups PEL + r XREADGROUP GROUP testxadgrp1 testxadcnsmr COUNT 1 STREAMS testxadstream > + r XREADGROUP GROUP testxadgrp2 testxadcnsmr COUNT 1 STREAMS testxadstream > + + # Group 1 + set pend [r XPENDING testxadstream testxadgrp1] + assert_equal 2-0 [lindex $pend 1] + set ids [r XACKDEL testxadstream testxadgrp1 IDS 1 2] + assert_equal 1 [llength $ids] + assert_equal 1 [lindex $ids 0] + assert_equal {} [r xrange testxadstream 2 2] + + # Group 2 still has ref in PEL + set pend [r XPENDING testxadstream testxadgrp2] + assert_equal 2-0 [lindex $pend 1] + } + + test {XACKDEL w/ ACKED doesn't delete when 2nd consumer group has message in PEL} { + r DEL testxadstream + r XADD testxadstream 1 msg hello + r XGROUP CREATE testxadstream testxadgrp1 1 + r XGROUP CREATE testxadstream testxadgrp2 1 + r XADD testxadstream 2 msg2 hello2 + + # Setup consumer groups w/ message in both groups PEL + r XREADGROUP GROUP testxadgrp1 testxadcnsmr COUNT 1 STREAMS testxadstream > + r XREADGROUP GROUP testxadgrp2 testxadcnsmr COUNT 1 STREAMS testxadstream > + + # Group 1 w/ ACKED only ack's, doesn't delete b/c group2 still hasn't gotten there + set pend [r XPENDING testxadstream testxadgrp1] + assert_equal 2-0 [lindex $pend 1] + set ids [r XACKDEL testxadstream testxadgrp1 ACKED IDS 1 2] + assert_equal 1 [llength $ids] + assert_equal 2 [lindex $ids 0] + assert_equal {{2-0 {msg2 hello2}}} [r xrange testxadstream 2 2] + + # Group 2 w/ ACKED both ack's and deletes now that all groups have ACK'd + set pend [r XPENDING testxadstream testxadgrp2] + assert_equal 2-0 [lindex $pend 1] + set ids [r XACKDEL testxadstream testxadgrp2 ACKED IDS 1 2] + assert_equal 1 [llength $ids] + assert_equal 1 [lindex $ids 0] + assert_equal {} [r xrange testxadstream 2 2] + } + + # The claim checking logic uses `last_id`. So using XCLAIM to FORCE setting the LAST_ID + # would naturally affect this. + test {XACKDEL w/ ACKED doesn't delete when 2nd consumer group hasn't claimed message yet} { + r DEL testxadstream + r XADD testxadstream 1 msg hello + r XGROUP CREATE testxadstream testxadgrp1 1 + r XGROUP CREATE testxadstream testxadgrp2 1 + r XADD testxadstream 2 msg2 hello2 + + # Group 1 w/ ACKED only ack's, doesn't delete b/c group2 still hasn't gotten there + r XREADGROUP GROUP testxadgrp1 testxadcnsmr COUNT 1 STREAMS testxadstream > + set pend [r XPENDING testxadstream testxadgrp1] + assert_equal 2-0 [lindex $pend 1] + set ids [r XACKDEL testxadstream testxadgrp1 ACKED IDS 1 2] + assert_equal 1 [llength $ids] + assert_equal 2 [lindex $ids 0] + assert_equal {{2-0 {msg2 hello2}}} [r xrange testxadstream 2 2] + + # Group 2 w/ ACKED both ack's and deletes now that all groups have ACK'd + r XREADGROUP GROUP testxadgrp2 testxadcnsmr COUNT 1 STREAMS testxadstream > + set ids [r XACKDEL testxadstream testxadgrp2 ACKED IDS 1 2-0] + assert_equal 1 [llength $ids] + assert_equal 1 [lindex $ids 0] + assert_equal {} [r xrange testxadstream 2 2] + } + + test {XACKDEL w/ ACKED acks dangling PEL reference after plain XDEL} { + r DEL testxadstream + r XADD testxadstream 1-0 msg hello + r XGROUP CREATE testxadstream testxadgrp1 0 + r XGROUP CREATE testxadstream testxadgrp2 0 + + # Only group 1 delivers the message; group 2's last_id stays at 0-0. + r XREADGROUP GROUP testxadgrp1 testxadcnsmr COUNT 1 STREAMS testxadstream > + set pend [r XPENDING testxadstream testxadgrp1] + assert_equal 1-0 [lindex $pend 1] + + # Plain XDEL removes the entry but leaves group 1's PEL reference dangling. + assert_equal 1 [r XDEL testxadstream 1-0] + + # Acking the dangling reference replies 1 (acked, nothing left to + # delete), not 2 (blocked by group 2): the entry no longer exists and + # can never be delivered to group 2. + set ids [r XACKDEL testxadstream testxadgrp1 ACKED IDS 1 1-0] + assert_equal 1 [llength $ids] + assert_equal 1 [lindex $ids 0] + + # The dangling reference is gone from group 1's PEL and the stream is + # still empty. + assert_equal 0 [r XLEN testxadstream] + assert_equal {} [lindex [r XPENDING testxadstream testxadgrp1] 1] + assert_equal {} [lindex [r XPENDING testxadstream testxadgrp2] 1] + } + + test {XACKDEL w/ ACKED doesn't delete when 2nd group has message pending after plain XDEL} { + r DEL testxadstream + r XADD testxadstream 1-0 msg hello + r XGROUP CREATE testxadstream testxadgrp1 0 + r XGROUP CREATE testxadstream testxadgrp2 0 + + # Both groups deliver the message, then plain XDEL removes the entry, + # leaving both PEL references dangling. + r XREADGROUP GROUP testxadgrp1 testxadcnsmr COUNT 1 STREAMS testxadstream > + r XREADGROUP GROUP testxadgrp2 testxadcnsmr COUNT 1 STREAMS testxadstream > + assert_equal 1 [r XDEL testxadstream 1-0] + + # Group 2 still has the message pending, so deletion stays blocked (2). + set ids [r XACKDEL testxadstream testxadgrp1 ACKED IDS 1 1-0] + assert_equal 1 [llength $ids] + assert_equal 2 [lindex $ids 0] + + # Group 1 was acked, group 2's reference is untouched. + assert_equal {} [lindex [r XPENDING testxadstream testxadgrp1] 1] + assert_equal 1-0 [lindex [r XPENDING testxadstream testxadgrp2] 1] + } + + test {XACKDEL w/ DELREF deletes from stream and 2nd consumer group's PEL even if not ACK'd} { + r DEL testxadstream + r XADD testxadstream 1 msg hello + r XGROUP CREATE testxadstream testxadgrp1 1 + r XGROUP CREATE testxadstream testxadgrp2 1 + r XADD testxadstream 2 msg2 hello2 + + # Setup consumer groups w/ message in both groups PEL + r XREADGROUP GROUP testxadgrp1 testxadcnsmr COUNT 1 STREAMS testxadstream > + r XREADGROUP GROUP testxadgrp2 testxadcnsmr COUNT 1 STREAMS testxadstream > + + # Group 1 w/ DELREF does ACK in group 1, removes from group 1's PEL, and deletes from stream + set pend [r XPENDING testxadstream testxadgrp1] + assert_equal 2-0 [lindex $pend 1] + set ids [r XACKDEL testxadstream testxadgrp1 DELREF IDS 1 2] + assert_equal 1 [llength $ids] + assert_equal 1 [lindex $ids 0] + assert_equal {} [r xrange testxadstream 2 2] + + # And Group 2 still has the message in it's PEL + set pend [r XPENDING testxadstream testxadgrp2] + assert_equal {} [lindex $pend 1] + } + + test {XACKDEL w/ DELREF skips deleting refs when target group never received message} { + r DEL testxadstream + r XADD testxadstream 1-0 msg hello + # grp1 created first and reads the message; grp2 (target) never reads it. + # "grp1" < "grp2" so grp1 is iterated first. + r XGROUP CREATE testxadstream grp1 0 + r XGROUP CREATE testxadstream grp2 0 + r XREADGROUP GROUP grp1 testxadcnsmr COUNT 1 STREAMS testxadstream > + + # Target grp2 never had 1-0 pending -> must reply -1 and not change any state. + set ids [r XACKDEL testxadstream grp2 DELREF IDS 1 1-0] + assert_equal 1 [llength $ids] + assert_equal -1 [lindex $ids 0] + + # Stream entry is still be present. + assert_equal {{1-0 {msg hello}}} [r XRANGE testxadstream 1-0 1-0] + + # grp1's PEL entry is untouched. + set pend [r XPENDING testxadstream grp1] + assert_equal 1-0 [lindex $pend 1] + } + + test {XACKDEL w/ ACKED is a no-op when target group already acked the message} { + r DEL testxadstream + r XADD testxadstream 1-0 msg hello + r XGROUP CREATE testxadstream grp1 0 + r XGROUP CREATE testxadstream grp2 0 + r XREADGROUP GROUP grp1 testxadcnsmr COUNT 1 STREAMS testxadstream > + r XREADGROUP GROUP grp2 testxadcnsmr COUNT 1 STREAMS testxadstream > + # Target grp2 acks the message, dropping it from grp2's PEL. + r XACK testxadstream grp2 1-0 + + # grp2 no longer has 1-0 pending, so reply -1 and dont modify anything. + set ids [r XACKDEL testxadstream grp2 ACKED IDS 1 1-0] + assert_equal 1 [llength $ids] + assert_equal -1 [lindex $ids 0] + assert_equal {{1-0 {msg hello}}} [r XRANGE testxadstream 1-0 1-0] + + # grp1 still holds its PEL entry. + set pend [r XPENDING testxadstream grp1] + assert_equal 1-0 [lindex $pend 1] + } + + test {XACKDEL w/ mix of existing and non-existent messages} { + r DEL testxadstream + r XADD testxadstream 1 msg hello + r XGROUP CREATE testxadstream testxadgrp1 1 + r XGROUP CREATE testxadstream testxadgrp2 1 + r XADD testxadstream 2 msg2 hello2 + + # Setup consumer groups w/ message in both groups PEL + r XREADGROUP GROUP testxadgrp1 testxadcnsmr COUNT 1 STREAMS testxadstream > + r XREADGROUP GROUP testxadgrp2 testxadcnsmr COUNT 1 STREAMS testxadstream > + + # Group 1 + set pend [r XPENDING testxadstream testxadgrp1] + assert_equal 2-0 [lindex $pend 1] + set ids [r XACKDEL testxadstream testxadgrp1 IDS 4 2 10 99 234] + assert_equal 4 [llength $ids] + assert_equal 1 [lindex $ids 0] + assert_equal -1 [lindex $ids 1] + assert_equal -1 [lindex $ids 2] + assert_equal -1 [lindex $ids 3] + assert_equal {} [r xrange testxadstream 2 2] + + # Group 2 still has ref in PEL + set pend [r XPENDING testxadstream testxadgrp2] + assert_equal 2-0 [lindex $pend 1] + } + + test {XACKDEL multiple IDs some acked some not} { + r DEL testxadstream + r XADD testxadstream 1-0 f v1 + r XADD testxadstream 2-0 f v2 + r XADD testxadstream 3-0 f v3 + r XGROUP CREATE testxadstream testxadgrp1 0 + r XREADGROUP GROUP testxadgrp1 testxadcnsmr COUNT 10 STREAMS testxadstream > + + # Delete 1-0 and 3-0; leave 2-0 in stream; 99-0 doesn't exist + set ids [r XACKDEL testxadstream testxadgrp1 ACKED IDS 3 1-0 3-0 99-0] + assert_equal 3 [llength $ids] + assert_equal 1 [lindex $ids 0] ;# 1-0 deleted + assert_equal 1 [lindex $ids 1] ;# 3-0 deleted + assert_equal -1 [lindex $ids 2] ;# 99-0 not found + assert_equal 1 [r XLEN testxadstream] + assert_equal 2-0 [lindex [lindex [r XRANGE testxadstream - +] 0] 0] + } + + test {XACKDEL w/ message not claimed does nothing} { + r DEL testxadstream + r XADD testxadstream 1 msg hello + r XGROUP CREATE testxadstream testxadgrp1 1 + r XGROUP CREATE testxadstream testxadgrp2 1 + r XADD testxadstream 2 msg2 hello2 + + # Group 1 + set ids [r XACKDEL testxadstream testxadgrp1 ACKED IDS 2 99 2] + assert_equal 2 [llength $ids] + assert_equal -1 [lindex $ids 0] + assert_equal -1 [lindex $ids 1] + assert_equal {{2-0 {msg2 hello2}}} [r xrange testxadstream 2 2] + + # Group 2 + set ids [r XACKDEL testxadstream testxadgrp2 ACKED IDS 2 2 99] + assert_equal 2 [llength $ids] + assert_equal -1 [lindex $ids 0] + assert_equal -1 [lindex $ids 1] + assert_equal {{2-0 {msg2 hello2}}} [r xrange testxadstream 2 2] + + # Check stream length + assert_equal 2 [r xlen testxadstream] + } + + test {XACKDEL run multiple times returns -1 after first time} { + r DEL testxadstream + r XADD testxadstream 1 msg hello + r XGROUP CREATE testxadstream testxadgrp1 1 + r XGROUP CREATE testxadstream testxadgrp2 1 + r XADD testxadstream 2 msg2 hello2 + + # Setup consumer groups w/ message in both groups PEL + r XREADGROUP GROUP testxadgrp1 testxadcnsmr COUNT 1 STREAMS testxadstream > + r XREADGROUP GROUP testxadgrp2 testxadcnsmr COUNT 1 STREAMS testxadstream > + + # Group 1 w/ ACKED only ack's, doesn't delete b/c group2 still hasn't gotten there + set pend [r XPENDING testxadstream testxadgrp1] + assert_equal 2-0 [lindex $pend 1] + set ids [r XACKDEL testxadstream testxadgrp1 ACKED IDS 1 2] + assert_equal 1 [llength $ids] + assert_equal 2 [lindex $ids 0] + assert_equal {{2-0 {msg2 hello2}}} [r xrange testxadstream 2 2] + + # Group 1 run again now returns -1 + set ids [r XACKDEL testxadstream testxadgrp1 ACKED IDS 1 2] + assert_equal 1 [llength $ids] + assert_equal -1 [lindex $ids 0] + assert_equal {{2-0 {msg2 hello2}}} [r xrange testxadstream 2 2] + + # Group 2 w/ ACKED both ack's and deletes now that all groups have ACK'd + set pend [r XPENDING testxadstream testxadgrp2] + assert_equal 2-0 [lindex $pend 1] + set ids [r XACKDEL testxadstream testxadgrp2 ACKED IDS 1 2] + assert_equal 1 [llength $ids] + assert_equal 1 [lindex $ids 0] + assert_equal {} [r xrange testxadstream 2 2] + } + + test {XACKDEL with non-existent stream and group} { + r DEL testxadstream + + # Missing stream and group + set ids [r XACKDEL testxadstream testxadgrp1 ACKED IDS 1 2] + assert_equal 1 [llength $ids] + assert_equal -1 [lindex $ids 0] + + # Missing Group + r XADD testxadstream 1 msg hello + set ids [r XACKDEL testxadstream testxadgrp1 ACKED IDS 1 2] + assert_equal 1 [llength $ids] + assert_equal -1 [lindex $ids 0] + } + + test {XACKDEL should fail if given an invalid stream ID} { + r DEL testxadstream + r XADD testxadstream 1 msg hello + r XGROUP CREATE testxadstream testxadgrp1 1 + r XREADGROUP GROUP testxadgrp1 testxadcnsmr COUNT 1 STREAMS testxadstream > + assert_error "*Invalid stream ID*" {r XACKDEL testxadstream testxadgrp1 IDS 1 not-a-valid-id} + } + + test {XACKDEL should fail if called on a non-stream key} { + r DEL testxadstream + r SET testxadstream notastream + assert_error "*WRONGTYPE*" {r XACKDEL testxadstream testxadgrp1 IDS 1 1-0} + r DEL testxadstream + } + + test {XACKDEL should fail if given an unrecognized mode} { + r DEL testxadstream + r XADD testxadstream 1 msg hello + r XGROUP CREATE testxadstream testxadgrp1 1 + assert_error "*" {r XACKDEL testxadstream testxadgrp1 BADMODE IDS 1 1-0} + } + + test {XACKDEL IDS numids must be a positive integer} { + r DEL testxadstream + r XADD testxadstream 1-0 f v + r XGROUP CREATE testxadstream testxadgrp1 0 + assert_error {*Number of IDs must be a positive integer*} {r XACKDEL testxadstream testxadgrp1 IDS abc 1-0} + assert_error {*Number of IDs must be a positive integer*} {r XACKDEL testxadstream testxadgrp1 IDS 0 1-0} + assert_error {*Number of IDs must be a positive integer*} {r XACKDEL testxadstream testxadgrp1 IDS -5 1-0} + } + + test {XACKDEL IDS numids must match argument count} { + r DEL testxadstream + r XADD testxadstream 1-0 f v + r XGROUP CREATE testxadstream testxadgrp1 0 + assert_error {*syntax error*} {r XACKDEL testxadstream testxadgrp1 IDS 3 1-0 2-0} + assert_error {*syntax error*} {r XACKDEL testxadstream testxadgrp1 IDS 1 1-0 2-0} + } + + test {XACKDEL with more than 8 IDs exercises dynamic allocation} { + r DEL teststream + # STREAMID_STATIC_VECTOR_LEN is 8, use 10 to force zmalloc path + for {set i 1} {$i <= 10} {incr i} { + r XADD teststream $i msg hello + } + r XGROUP CREATE teststream testgrp 0 + r XREADGROUP GROUP testgrp consumer1 COUNT 10 STREAMS teststream > + set ids [r XACKDEL teststream testgrp IDS 10 1 2 3 4 5 6 7 8 9 10] + assert_equal 10 [llength $ids] + foreach id $ids { + assert_equal 1 $id + } + assert_equal 0 [r XLEN teststream] + } + + test {XACKDEL returns -1 for deleted entry not in group PEL} { + r DEL teststream + r XADD teststream 1-0 msg hello + r XGROUP CREATE teststream testgrp 0 + # Message never claimed, so not in any PEL. Delete it from the stream. + r XDEL teststream 1-0 + set res [r XACKDEL teststream testgrp IDS 1 1-0] + assert_equal 1 [llength $res] + assert_equal -1 [lindex $res 0] + } + + test {XACKDEL updates first_id via streamGetEdgeID when first entry is deleted but stream is non-empty} { + r DEL testxadstream + r XADD testxadstream 1-0 msg hello + r XADD testxadstream 2-0 msg2 hello2 + r XGROUP CREATE testxadstream testxadgrp1 0 + r XREADGROUP GROUP testxadgrp1 testxadcnsmr COUNT 10 STREAMS testxadstream > + + # Delete 1-0 (the first entry) while 2-0 remains; triggers the streamGetEdgeID path + set ids [r XACKDEL testxadstream testxadgrp1 ACKED IDS 1 1-0] + assert_equal 1 [llength $ids] + assert_equal 1 [lindex $ids 0] + + # 2-0 must still be present and must now be the first entry (first_id updated) + assert_equal 1 [r XLEN testxadstream] + set entries [r XRANGE testxadstream - +] + assert_equal 1 [llength $entries] + assert_equal 2-0 [lindex [lindex $entries 0] 0] + } + + test {XACKDEL DELREF deletes entry when non-target group has not yet claimed it} { + r DEL testxadstream + r XADD testxadstream 1-0 msg hello + r XGROUP CREATE testxadstream testxadgrp1 0 + r XGROUP CREATE testxadstream testxadgrp2 0 + + # Only grp1 claims the message; grp2 has never read it (beyond grp2's last_id) + r XREADGROUP GROUP testxadgrp1 testxadcnsmr COUNT 1 STREAMS testxadstream > + + set ids [r XACKDEL testxadstream testxadgrp1 DELREF IDS 1 1-0] + assert_equal 1 [llength $ids] + assert_equal 1 [lindex $ids 0] + + # Entry must be deleted from the stream even though grp2 never had a PEL entry for it + assert_equal 0 [r XLEN testxadstream] + assert_equal {} [r XRANGE testxadstream - +] + set pend2 [r XPENDING testxadstream testxadgrp2] + assert_equal 0 [lindex $pend2 0] + } + + test {XACKDEL ACKED returns 1 when stream entry was already removed via XDEL} { + r DEL testxadstream + r XADD testxadstream 1-0 f v + r XGROUP CREATE testxadstream testxadgrp1 0 + r XREADGROUP GROUP testxadgrp1 testxadcnsmr COUNT 1 STREAMS testxadstream > + + # XDEL removes the stream entry but leaves the PEL entry intact + r XDEL testxadstream 1-0 + set pend [r XPENDING testxadstream testxadgrp1] + assert_equal 1 [lindex $pend 0] + + # XACKDEL clears the PEL and still returns 1 even though already deleted + set ids [r XACKDEL testxadstream testxadgrp1 ACKED IDS 1 1-0] + assert_equal 1 [llength $ids] + assert_equal 1 [lindex $ids 0] + + # Check that XACKDEL ACKED removes the dangling PEL entry left after XDEL + set pend [r XPENDING testxadstream testxadgrp1] + assert_equal 0 [lindex $pend 0] + } + + test {XACKDEL ACKED with a single consumer group deletes the entry} { + r DEL testxadstream + r XADD testxadstream 1-0 f v + r XGROUP CREATE testxadstream testxadgrp1 0 + r XREADGROUP GROUP testxadgrp1 testxadcnsmr COUNT 1 STREAMS testxadstream > + + set ids [r XACKDEL testxadstream testxadgrp1 ACKED IDS 1 1-0] + assert_equal 1 [llength $ids] + assert_equal 1 [lindex $ids 0] + assert_equal 0 [r XLEN testxadstream] + set pend [r XPENDING testxadstream testxadgrp1] + assert_equal 0 [lindex $pend 0] + } + + test {XACKDEL ACKED drains PEL despite XGROUP SETID moving last_id backward} { + r DEL testxadstream + r XADD testxadstream 1-0 f v1 + r XADD testxadstream 5-0 f v5 + r XGROUP CREATE testxadstream testxadgrp1 0 + + # Claim both entries: last_id advances to 5-0 and both land in the PEL. + r XREADGROUP GROUP testxadgrp1 testxadcnsmr COUNT 10 STREAMS testxadstream > + set pend [r XPENDING testxadstream testxadgrp1] + assert_equal 2 [lindex $pend 0] + assert_equal 5-0 [lindex $pend 2] + + # Move last_id back before the latest claimed entry. 5-0 stays in the + # PEL even though id > last_id now. + r XGROUP SETID testxadstream testxadgrp1 1-0 + set pend [r XPENDING testxadstream testxadgrp1] + assert_equal 2 [lindex $pend 0] + + # ACKED must still drain the dangling 5-0 PEL entry and delete the msg. + set ids [r XACKDEL testxadstream testxadgrp1 ACKED IDS 1 5-0] + assert_equal 1 [llength $ids] + assert_equal 1 [lindex $ids 0] + assert_equal {} [r xrange testxadstream 5-0 5-0] + + # 5-0 should be gone from the PEL; 1-0 remains pending. + set pend [r XPENDING testxadstream testxadgrp1] + assert_equal 1 [lindex $pend 0] + assert_equal 1-0 [lindex $pend 1] + } + + test {XDELEX deletes items} { + r DEL teststream + r XADD teststream 1 msg helllo + r XADD teststream 2 msg helllo + r XADD teststream 3 msg helllo + set resp [ r XDELEX teststream IDS 1 1 ] + assert_equal 1 [llength $resp] + assert_equal 1 [lindex $resp 0] + + assert_equal 2 [ r XLEN teststream ] + } + + test {XDELEX on non-existent key returns -1 for each ID} { + r DEL nonexistent + set resp [r XDELEX nonexistent IDS 2 1 2] + assert_equal 2 [llength $resp] + assert_equal -1 [lindex $resp 0] + assert_equal -1 [lindex $resp 1] + } + + test {XDELEX on wrong key type returns error} { + r DEL testset + r SADD testset a b c + assert_error "*WRONGTYPE*" {r XDELEX testset IDS 1 1} + } + + test {XDELEX w/ ACKED only deletes acked items} { + r DEL teststream + r XADD teststream 1 msg helllo + r XADD teststream 2 msg helllo + r XADD teststream 3 msg helllo + + r XGROUP CREATE teststream testgrp1 0 + r XREADGROUP GROUP testgrp1 testconsumer COUNT 1 STREAMS teststream > + r XACK teststream testgrp1 1 + + set ids [ r XDELEX teststream ACKED IDS 3 1 2 99 ] + assert_equal 3 [llength $ids] + assert_equal 1 [lindex $ids 0] + assert_equal 2 [lindex $ids 1] + assert_equal -1 [lindex $ids 2] + + assert_equal 2 [ r XLEN teststream ] + } + + test {XDELEX ACKED blocks deletion for entries pending below a rewound last_id} { + r DEL testxadstream + r XADD testxadstream 1-0 f v1 + r XADD testxadstream 5-0 f v5 + r XGROUP CREATE testxadstream testxadgrp1 0 + + # Claim both entries: last_id advances to 5-0 and both land in the PEL. + r XREADGROUP GROUP testxadgrp1 testxadcnsmr COUNT 10 STREAMS testxadstream > + + # Move last_id back before 5-0, which stays in the PEL (id > last_id). + r XGROUP SETID testxadstream testxadgrp1 1-0 + + # 5-0 is still pending in the group, so ACKED must not delete it. + set ids [r XDELEX testxadstream ACKED IDS 1 5-0] + assert_equal 2 [lindex $ids 0] + assert_equal 2 [r XLEN testxadstream] + set pend [r XPENDING testxadstream testxadgrp1] + assert_equal 2 [lindex $pend 0] + + # After XACK drains the PEL entry, deletion stays blocked because the + # rewound delivery cursor means the group may still be served 5-0 by + # XREADGROUP "" (same fallback as XACKDEL ACKED). + r XACK testxadstream testxadgrp1 5-0 + set ids [r XDELEX testxadstream ACKED IDS 1 5-0] + assert_equal 2 [lindex $ids 0] + assert_equal 2 [r XLEN testxadstream] + + # Catching the cursor back up unblocks deletion. + r XGROUP SETID testxadstream testxadgrp1 5-0 + set ids [r XDELEX testxadstream ACKED IDS 1 5-0] + assert_equal 1 [lindex $ids 0] + assert_equal 1 [r XLEN testxadstream] + } + + test {XDELEX ACKED reports pending ref even when a clean group is iterated first} { + r DEL testxadstream + r XADD testxadstream 1-0 f v1 + r XADD testxadstream 5-0 f v5 + # agrp sorts before zgrp, so it is iterated first; it never claims 5-0. + r XGROUP CREATE testxadstream agrp 0 + r XGROUP CREATE testxadstream zgrp 0 + r XREADGROUP GROUP zgrp zcnsmr COUNT 10 STREAMS testxadstream > + + # Remove the stream entry, leaving 5-0 dangling only in zgrp's PEL. + r XDEL testxadstream 5-0 + + # The dangling pending ref must block ACKED with 2; the reply cannot + # depend on which consumer group the iteration reaches first. + set ids [r XDELEX testxadstream ACKED IDS 1 5-0] + assert_equal 2 [lindex $ids 0] + assert_equal 1 [r XLEN testxadstream] + set pend [r XPENDING testxadstream zgrp] + assert_equal 2 [lindex $pend 0] + } + + test {XDELEX ACKED returns -1 for deleted entry not referenced by any group} { + r DEL testxadstream + r XADD testxadstream 1-0 f v1 + r XADD testxadstream 5-0 f v5 + r XGROUP CREATE testxadstream testxadgrp1 0 + r XREADGROUP GROUP testxadgrp1 testxadcnsmr COUNT 10 STREAMS testxadstream > + r XGROUP SETID testxadstream testxadgrp1 1-0 + r XACK testxadstream testxadgrp1 5-0 + + # Remove the entry; nothing references it anymore. The rewound last_id + # must not produce 2, since a deleted entry can never be re-delivered. + r XDEL testxadstream 5-0 + set ids [r XDELEX testxadstream ACKED IDS 1 5-0] + assert_equal -1 [lindex $ids 0] + assert_equal 1 [r XLEN testxadstream] + } + + test {XDELEX ACKED reports pending ref, not -1, when entry is deleted below a rewound last_id} { + r DEL testxadstream + r XADD testxadstream 1-0 f v1 + r XADD testxadstream 5-0 f v5 + r XGROUP CREATE testxadstream testxadgrp1 0 + + # Claim both entries, then rewind last_id below the still-pending 5-0. + r XREADGROUP GROUP testxadgrp1 testxadcnsmr COUNT 10 STREAMS testxadstream > + r XGROUP SETID testxadstream testxadgrp1 1-0 + + # Remove the stream entry, leaving 5-0 dangling in the PEL. + r XDEL testxadstream 5-0 + + # The dangling pending ref must block ACKED (2), like when the group + # cursor is ahead of the ID, instead of reporting -1 (not found). + set ids [r XDELEX testxadstream ACKED IDS 1 5-0] + assert_equal 2 [lindex $ids 0] + assert_equal 1 [r XLEN testxadstream] + set pend [r XPENDING testxadstream testxadgrp1] + assert_equal 2 [lindex $pend 0] + } + + test {XDELEX DELREF clears PEL ref even when last_id was rewound below it} { + r DEL testxadstream + r XADD testxadstream 1-0 f v1 + r XADD testxadstream 5-0 f v5 + r XGROUP CREATE testxadstream testxadgrp1 0 + + # Claim both entries, then rewind last_id below the still-pending 5-0. + r XREADGROUP GROUP testxadgrp1 testxadcnsmr COUNT 10 STREAMS testxadstream > + r XGROUP SETID testxadstream testxadgrp1 1-0 + + # DELREF must remove the pending ref even though 5-0 > last_id. + set ids [r XDELEX testxadstream DELREF IDS 1 5-0] + assert_equal 1 [lindex $ids 0] + assert_equal 1 [r XLEN testxadstream] + + # Only 1-0 remains pending; the 5-0 ref must be gone. + set pend [r XPENDING testxadstream testxadgrp1] + assert_equal 1 [lindex $pend 0] + assert_equal 1-0 [lindex $pend 1] + assert_equal 1-0 [lindex $pend 2] + } + + test {XDELEX w/ KEEPREF deletes all but keeps refs in consumer group PELs} { + r DEL teststream + r XADD teststream 1 msg helllo + r XADD teststream 2 msg helllo + r XADD teststream 3 msg helllo + + r XGROUP CREATE teststream testgrp1 0 + r XREADGROUP GROUP testgrp1 testconsumer COUNT 1 STREAMS teststream > + + set ids [ r XDELEX teststream KEEPREF IDS 3 1 2 99 ] + assert_equal 3 [llength $ids] + assert_equal 1 [lindex $ids 0] + assert_equal 1 [lindex $ids 1] + assert_equal -1 [lindex $ids 2] + + assert_equal {1 1-0 1-0 {{testconsumer 1}}} [r XPENDING teststream testgrp1] + + assert_equal 1 [ r XLEN teststream ] + } + + test {XDELEX w/ DELREF deletes entries and clears consumer group PEL refs} { + r DEL teststream + r XADD teststream 1 msg helllo + r XADD teststream 2 msg helllo + r XADD teststream 3 msg helllo + + r XGROUP CREATE teststream testgrp1 0 + r XREADGROUP GROUP testgrp1 testconsumer COUNT 1 STREAMS teststream > + + set ids [ r XDELEX teststream DELREF IDS 3 1 2 99 ] + assert_equal 3 [llength $ids] + assert_equal 1 [lindex $ids 0] + assert_equal 1 [lindex $ids 1] + assert_equal -1 [lindex $ids 2] + + assert_equal {0 {} {} {}} [r XPENDING teststream testgrp1] + + assert_equal 1 [ r XLEN teststream ] + } + + test {XDELEX DELREF signals WATCH when removing an orphaned PEL ref} { + r DEL teststream + r XADD teststream 1-0 msg hello + r XGROUP CREATE teststream grp1 0 + r XREADGROUP GROUP grp1 consumer COUNT 1 STREAMS teststream > + + # Delete the entry but keep its (now orphaned) PEL reference. + r XDELEX teststream KEEPREF IDS 1 1-0 + assert_equal 0 [r XLEN teststream] + assert_equal {1 1-0 1-0 {{consumer 1}}} [r XPENDING teststream grp1] + + # DELREF clearing the orphaned NACK modifies the key even though no + # stream entry is deleted, so the pending MULTI/EXEC must abort and + # return the empty reply. + r WATCH teststream + r XDELEX teststream DELREF IDS 1 1-0 + r MULTI + r ping + assert_equal {} [r EXEC] + + assert_equal {0 {} {} {}} [r XPENDING teststream grp1] + } + + test {XDELEX DELREF removing an orphaned PEL ref emits an xdel keyspace event} { + r DEL teststream + r XADD teststream 1-0 msg hello + r XGROUP CREATE teststream grp1 0 + r XREADGROUP GROUP grp1 consumer COUNT 1 STREAMS teststream > + r XDELEX teststream KEEPREF IDS 1 1-0 + assert_equal {1 1-0 1-0 {{consumer 1}}} [r XPENDING teststream grp1] + + # Enable keyspace notifications: K = keyspace events on __keyspace@:, + # t = stream events (so xdel/XPENDING PEL mutations surface as 'xdel'). + r config set notify-keyspace-events Kt + + # Subscribe with a glob so the channel matches regardless of which DB + # tests are running under. The keyspace channel is __keyspace@: + # so * matches all db numbers. + set rd1 [valkey_deferring_client] + set subpat __keyspace@*:teststream + assert_equal {1} [psubscribe $rd1 $subpat] + + # No stream entry is deleted here, but the PEL change still notifies. + r XDELEX teststream DELREF IDS 1 1-0 + + # The pmessage reply is: pmessage . + # Here, is the xdel event we're looking for. The subscribed + # pattern is the glob from above, and the channel is also wildcarded by + # assert_match to avoid failing depending on the DB number in use for + # tests. + assert_match "pmessage $subpat __keyspace@*:teststream xdel" [$rd1 read] + + $rd1 close + assert_equal {0 {} {} {}} [r XPENDING teststream grp1] + r config set notify-keyspace-events {} + } + + test {XDELEX on already-deleted ID returns -1} { + r DEL teststream + r XADD teststream 1 msg hello + r XADD teststream 2 msg hello + + # First delete succeeds + set resp [r XDELEX teststream IDS 1 1] + assert_equal 1 [lindex $resp 0] + + # Second delete on the same ID returns -1 + set resp [r XDELEX teststream IDS 1 1] + assert_equal -1 [lindex $resp 0] + } + + test {XDELEX should fail on invalid stream ID format} { + r DEL teststream + r XADD teststream 1 msg hello + assert_error "*Invalid stream ID specified*" {r XDELEX teststream IDS 1 not-a-valid-id} + } + + test {XDELEX w/ KEEPREF preserves second consumer group's PEL} { + r DEL teststream + r XADD teststream 1-0 msg hello + r XADD teststream 2-0 msg hello + r XGROUP CREATE teststream grp1 0 + r XGROUP CREATE teststream grp2 0 + r XREADGROUP GROUP grp1 consumer COUNT 1 STREAMS teststream > + r XREADGROUP GROUP grp2 consumer COUNT 1 STREAMS teststream > + + # KEEPREF: delete from stream but leave grp2's PEL reference intact + set ids [r XDELEX teststream KEEPREF IDS 1 1-0] + assert_equal 1 [lindex $ids 0] + + assert_equal {1 1-0 1-0 {{consumer 1}}} [r XPENDING teststream grp2] + assert_equal 1 [r XLEN teststream] + } + + test {XDELEX w/ DELREF clears all consumer groups' PELs} { + r DEL teststream + r XADD teststream 1-0 msg hello + r XADD teststream 2-0 msg hello + r XGROUP CREATE teststream grp1 0 + r XGROUP CREATE teststream grp2 0 + r XREADGROUP GROUP grp1 consumer COUNT 1 STREAMS teststream > + r XREADGROUP GROUP grp2 consumer COUNT 1 STREAMS teststream > + + # DELREF: delete from stream and wipe all groups' PEL references + set ids [r XDELEX teststream DELREF IDS 1 1-0] + assert_equal 1 [lindex $ids 0] + + assert_equal {0 {} {} {}} [r XPENDING teststream grp1] + assert_equal {0 {} {} {}} [r XPENDING teststream grp2] + assert_equal 1 [r XLEN teststream] + } + + test {XDELEX w/ ACKED waits for all consumer groups before deleting} { + r DEL teststream + r XADD teststream 1-0 msg hello + r XGROUP CREATE teststream grp1 0 + r XGROUP CREATE teststream grp2 0 + r XREADGROUP GROUP grp1 consumer COUNT 1 STREAMS teststream > + r XREADGROUP GROUP grp2 consumer COUNT 1 STREAMS teststream > + + # grp1 acks but grp2 still has it in PEL → no deletion + r XACK teststream grp1 1-0 + set ids [r XDELEX teststream ACKED IDS 1 1-0] + assert_equal 2 [lindex $ids 0] + assert_equal 1 [r XLEN teststream] + + # grp2 acks; all groups done → deletion occurs + r XACK teststream grp2 1-0 + set ids [r XDELEX teststream ACKED IDS 1 1-0] + assert_equal 1 [lindex $ids 0] + assert_equal 0 [r XLEN teststream] + } + + test {XDELEX returns syntax error when IDS token is missing after mode} { + r DEL teststream + r XADD teststream 1 msg hello + assert_error "*syntax error*" {r XDELEX teststream KEEPREF NOIDS 1 1} + assert_error "*syntax error*" {r XDELEX teststream DELREF NOIDS 1 1} + assert_error "*syntax error*" {r XDELEX teststream ACKED NOIDS 1 1} + } + + test {XDELEX returns error for unrecognized mode token} { + r DEL teststream + r XADD teststream 1 msg hello + assert_error "*syntax error*" {r XDELEX teststream BADMODE IDS 1 1} + } + + test {XDELEX returns error for invalid numids} { + r DEL teststream + r XADD teststream 1 msg hello + assert_error "*positive integer*" {r XDELEX teststream IDS 0 1} + assert_error "*positive integer*" {r XDELEX teststream IDS -1 1} + assert_error "*positive integer*" {r XDELEX teststream IDS abc 1} + } + + test {XDELEX returns error when numids does not match ID count} { + r DEL teststream + r XADD teststream 1 msg hello + r XADD teststream 2 msg hello + # Too few IDs provided for numids + assert_error "*syntax error*" {r XDELEX teststream IDS 3 1 2} + # Too many IDs provided for numids + assert_error "*syntax error*" {r XDELEX teststream IDS 1 1 2} + } + + test {XDELEX with more than 8 IDs exercises dynamic allocation} { + r DEL teststream + # STREAMID_STATIC_VECTOR_LEN is 8, use 10 to force zmalloc path + for {set i 1} {$i <= 10} {incr i} { + r XADD teststream $i msg hello + } + set ids [r XDELEX teststream IDS 10 1 2 3 4 5 6 7 8 9 10] + assert_equal 10 [llength $ids] + foreach id $ids { + assert_equal 1 $id + } + assert_equal 0 [r XLEN teststream] + } + + test {XDELEX with more than 8 IDs and ACKED mode exercises dynamic allocation} { + r DEL teststream + for {set i 1} {$i <= 10} {incr i} { + r XADD teststream $i msg hello + } + r XGROUP CREATE teststream grp 0 + # Read all entries into the group's PEL + r XREADGROUP GROUP grp consumer COUNT 10 STREAMS teststream > + # Ack all so ACKED mode can delete them + for {set i 1} {$i <= 10} {incr i} { + r XACK teststream grp $i + } + set ids [r XDELEX teststream ACKED IDS 10 1 2 3 4 5 6 7 8 9 10] + assert_equal 10 [llength $ids] + foreach id $ids { + assert_equal 1 $id + } + assert_equal 0 [r XLEN teststream] + } + test {XRANGE fuzzing} { set items [r XRANGE mystream{t} - +] set low_id [lindex $items 0 0]