diff --git a/content/commands/xackdel.md b/content/commands/xackdel.md new file mode 100644 index 0000000000..bdbda003fd --- /dev/null +++ b/content/commands/xackdel.md @@ -0,0 +1,163 @@ +--- +acl_categories: +- '@write' +- '@stream' +- '@fast' +arguments: +- display_text: key + key_spec_index: 0 + name: key + type: key +- display_text: group + name: group + type: string +- arguments: + - display_text: keepref + name: keepref + token: KEEPREF + type: pure-token + - display_text: delref + name: delref + token: DELREF + type: pure-token + - display_text: acked + name: acked + token: ACKED + type: pure-token + name: condition + optional: true + type: oneof +- arguments: + - display_text: numids + name: numids + type: integer + - display_text: id + multiple: true + name: id + type: string + name: ids + token: IDS + type: block +arity: -6 +categories: +- docs +- develop +- stack +- oss +- rs +- rc +- oss +- kubernetes +- clients +command_flags: +- write +- fast +complexity: O(1) for each entry ID processed. +description: Acknowledges and conditionally deletes one or multiple entries for a stream consumer + group. +group: stream +hidden: false +key_specs: +- RW: true + begin_search: + spec: + index: 1 + type: index + delete: true + find_keys: + spec: + keystep: 1 + lastkey: 0 + limit: 0 + type: range + update: true +linkTitle: XACKDEL +since: 8.2.0 +summary: Acknowledges and conditionally deletes one or multiple entries for a stream consumer group. +syntax_fmt: "XACKDEL key group [KEEPREF | DELREF | ACKED] IDS\_numids id [id ...]" +syntax_str: "group [KEEPREF | DELREF | ACKED] IDS\_numids id [id ...]" +title: XACKDEL +--- + +Acknowledges and conditionally deletes one or multiple entries (messages) for a stream consumer group at the specified `key`. + +`XACKDEL` combines the functionality of [`XACK`]({{< relref "/commands/xack" >}}) and [`XDEL`]({{< relref "/commands/xdel" >}}) in Redis Streams. It acknowledges the specified entry IDs in the given consumer group and simultaneously attempts to delete the corresponding entries from the stream. + +## Required arguments + +
+key + +The name of the stream key. +
+ +
+group + +The name of the consumer group. +
+ +
+IDS numids id [id ...] + +The IDS block specifying which entries to acknowledge and delete: +- `numids`: The number of IDs that follow +- `id [id ...]`: One or more stream entry IDs to acknowledge and delete +
+ +## Optional arguments + +
+KEEPREF | DELREF | ACKED + +Specifies how to handle consumer group references when acknowledging and deleting entries. Available since Redis 8.2. If no option is specified, `KEEPREF` is used by default: + +- `KEEPREF` (default): Acknowledges the entries in the specified consumer group and deletes the entries from the stream, but preserves existing references to these entries in all consumer groups' PEL (Pending Entries List). +- `DELREF`: Acknowledges the entries in the specified consumer group, deletes the entries from the stream, and also removes all references to these entries from all consumer groups' pending entry lists, effectively cleaning up all traces of the entries. If an entry ID is not in the stream, but there are dangling references, `XACKDEL` with `DELREF` would still remove all those references. +- `ACKED`: Acknowledges the entries in the specified consumer group and only deletes entries that were read and acknowledged by all consumer groups. +
+ +This command is particularly useful when you want to both acknowledge entry processing and clean up the stream in a single atomic operation, providing fine-grained control over how entry references are handled. + +{{< note >}} +When using multiple consumer groups, users are encouraged to use `XACKDEL` with the `ACKED` option instead of `XACK` and `XDEL`, simplifying the application logic. +{{< /note >}} + +## Examples + +{{% redis-cli %}} +XADD mystream * field1 value1 +XADD mystream * field2 value2 +XGROUP CREATE mystream mygroup 0 +XREADGROUP GROUP mygroup consumer1 COUNT 2 STREAMS mystream > +XPENDING mystream mygroup +XACKDEL mystream mygroup KEEPREF IDS 2 1526919030474-55 1526919030474-56 +XPENDING mystream mygroup +XRANGE mystream - + +{{% /redis-cli %}} + +## Return information + +{{< multitabs id="xackdel-return-info" + tab1="RESP2" + tab2="RESP3" >}} + +One of the following: + +* [Array reply]({{< relref "/develop/reference/protocol-spec#arrays" >}}): -1 for each requested ID when the given key does not exist. +* [Array reply]({{< relref "/develop/reference/protocol-spec#arrays" >}}): For each ID: + * [Integer reply]({{< relref "/develop/reference/protocol-spec#integers" >}}): 1 if the entry was acknowledged and deleted from the stream. + * [Integer reply]({{< relref "/develop/reference/protocol-spec#integers" >}}): -1 if no such ID exists in the provided stream key. + * [Integer reply]({{< relref "/develop/reference/protocol-spec#integers" >}}): 2 if the entry was acknowledged but not deleted, as there are still dangling references (ACKED option). + +-tab-sep- + +One of the following: + +* [Array reply]({{< relref "/develop/reference/protocol-spec#arrays" >}}): -1 for each requested ID when the given key does not exist. +* [Array reply]({{< relref "/develop/reference/protocol-spec#arrays" >}}): For each ID: + * [Integer reply]({{< relref "/develop/reference/protocol-spec#integers" >}}): 1 if the entry was acknowledged and deleted from the stream. + * [Integer reply]({{< relref "/develop/reference/protocol-spec#integers" >}}): -1 if no such ID exists in the provided stream key. + * [Integer reply]({{< relref "/develop/reference/protocol-spec#integers" >}}): 2 if the entry was acknowledged but not deleted, as there are still dangling references (ACKED option). + +{{< /multitabs >}} diff --git a/content/commands/xadd.md b/content/commands/xadd.md index 363527f12b..b45e2dd906 100644 --- a/content/commands/xadd.md +++ b/content/commands/xadd.md @@ -14,6 +14,22 @@ arguments: since: 6.2.0 token: NOMKSTREAM type: pure-token +- arguments: + - display_text: keepref + name: keepref + token: KEEPREF + type: pure-token + - display_text: delref + name: delref + token: DELREF + type: pure-token + - display_text: acked + name: acked + token: ACKED + type: pure-token + name: condition + optional: true + type: oneof - arguments: - arguments: - display_text: maxlen @@ -98,6 +114,8 @@ history: - 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. key_specs: - RW: true begin_search: @@ -115,44 +133,93 @@ key_specs: linkTitle: XADD since: 5.0.0 summary: Appends a new message to a stream. Creates the key if it doesn't exist. -syntax_fmt: "XADD key [NOMKSTREAM] [ [= | ~] threshold\n [LIMIT\_\ - count]] <* | id> field value [field value ...]" -syntax_str: "[NOMKSTREAM] [ [= | ~] threshold [LIMIT\_count]] <* |\ - \ id> field value [field value ...]" +syntax_fmt: "XADD key [NOMKSTREAM] [KEEPREF | DELREF | ACKED] [\n\ + \ [= | ~] threshold [LIMIT\_count]] <* | id> field value [field value\n ...]" +syntax_str: "[NOMKSTREAM] [KEEPREF | DELREF | ACKED] [ [= | ~] threshold\ + \ [LIMIT\_count]] <* | id> field value [field value ...]" title: XADD --- -Appends the specified stream entry to the stream at the specified key. -If the key does not exist, as a side effect of running this command the -key is created with a stream value. The creation of stream's key can be -disabled with the `NOMKSTREAM` option. -An entry is composed of a list of field-value pairs. -The field-value pairs are stored in the same order they are given by the user. -Commands that read the stream, such as [`XRANGE`]({{< relref "/commands/xrange" >}}) or [`XREAD`]({{< relref "/commands/xread" >}}), are guaranteed to return the fields and values exactly in the same order they were added by `XADD`. +Appends the specified stream entry to the stream at the specified `key`. +If the key does not exist, `XADD` will create a new key with the given stream value as a side effect of running this command. +You can turn off key creation with the `NOMKSTREAM` option. + +## Required arguments + +
+key + +The name of the stream key. +
+ +
+id + +The stream entry ID. Use `*` to auto-generate a unique ID, or specify a well-formed ID in the format `-` (for example, `1526919030474-55`). +
+ +
+field value [field value ...] -`XADD` is the *only Redis command* that can add data to a stream, but -there are other commands, such as [`XDEL`]({{< relref "/commands/xdel" >}}) and [`XTRIM`]({{< relref "/commands/xtrim" >}}), that are able to +One or more field-value pairs that make up the stream entry. You must provide at least one field-value pair. +
+ +## Optional arguments + +
+NOMKSTREAM + +Prevents the creation of a new stream if the key does not exist. Available since Redis 6.2.0. +
+ +
+KEEPREF | DELREF | ACKED + +Specifies how to handle consumer group references when trimming. Available since Redis 8.2. If no option is specified, `KEEPREF` is used by default. Unlike the `XDELEX` and `XACKDEL` commands where one of these options is required, here they are optional to maintain backward compatibility: + +- `KEEPREF` (default): When trimming, removes entries from the stream according to the specified strategy (`MAXLEN` or `MINID`), regardless of whether they are referenced by any consumer groups, but preserves existing references to these entries in all consumer groups' PEL (Pending Entries List). +- `DELREF`: When trimming, removes entries from the stream according to the specified strategy and also removes all references to these entries from all consumer groups' PEL. +- `ACKED`: When trimming, only removes entries that were read and acknowledged by all consumer groups. Note that if the number of referenced entries is larger than `MAXLEN`, trimming will still stop at the limit. +
+ +
+MAXLEN | MINID [= | ~] threshold [LIMIT count] + +Trims the stream to maintain a specific size or remove old entries: +- `MAXLEN`: Limits the stream to a maximum number of entries +- `MINID`: Removes entries with IDs lower than the specified threshold (available since Redis 6.2.0) +- `=`: Exact trimming (default) +- `~`: Approximate trimming (more efficient) +- `threshold`: The maximum number of entries (for MAXLEN) or minimum ID (for MINID) +- `LIMIT count`: Limits the number of entries to examine during trimming (available since Redis 6.2.0) +
+ +Each entry consists of a list of field-value pairs. +Redis stores the field-value pairs in the same order you provide them. +Commands that read the stream, such as [`XRANGE`]({{< relref "/commands/xrange" >}}) or [`XREAD`]({{< relref "/commands/xread" >}}), return the fields and values in exactly the same order you added them with `XADD`. + +{{< note >}} +`XADD` is the only Redis command that can add data to a stream. However, +other commands, such as [`XDEL`]({{< relref "/commands/xdel" >}}) and [`XTRIM`]({{< relref "/commands/xtrim" >}}), can remove data from a stream. +{{< /note >}} ## Specifying a Stream ID as an argument -A stream entry ID identifies a given entry inside a stream. +A stream entry ID identifies a specific entry inside a stream. -The `XADD` command will auto-generate a unique ID for you if the ID argument -specified is the `*` character (asterisk ASCII character). However, while -useful only in very rare cases, it is possible to specify a well-formed ID, so -that the new entry will be added exactly with the specified ID. +`XADD` auto-generates a unique ID for you if you specify the `*` character (asterisk) as the ID argument. However, you can also specify a well-formed ID to add the new entry with that exact ID, though this is useful only in rare cases. -IDs are specified by two numbers separated by a `-` character: +Specify IDs using two numbers separated by a `-` character: 1526919030474-55 -Both quantities are 64-bit numbers. When an ID is auto-generated, the +Both numbers are 64-bit integers. When Redis auto-generates an ID, the first part is the Unix time in milliseconds of the Redis instance generating -the ID. The second part is just a sequence number and is used in order to +the ID. The second part is a sequence number used to distinguish IDs generated in the same millisecond. -You can also specify an incomplete ID, that consists only of the milliseconds part, which is interpreted as a zero value for sequence part. +You can also specify an incomplete ID that consists only of the milliseconds part, which Redis interprets as a zero value for the sequence part. To have only the sequence part automatically generated, specify the milliseconds part followed by the `-` separator and the `*` character: ``` @@ -162,37 +229,25 @@ To have only the sequence part automatically generated, specify the milliseconds "1526919030474-56" ``` -IDs are guaranteed to be always incremental: If you compare the ID of the -entry just inserted it will be greater than any other past ID, so entries -are totally ordered inside a stream. In order to guarantee this property, -if the current top ID in the stream has a time greater than the current -local time of the instance, the top entry time will be used instead, and -the sequence part of the ID incremented. This may happen when, for instance, -the local clock jumps backward, or if after a failover the new master has -a different absolute time. - -When a user specified an explicit ID to `XADD`, the minimum valid ID is -`0-1`, and the user *must* specify an ID which is greater than any other -ID currently inside the stream, otherwise the command will fail and return an error. Usually -resorting to specific IDs is useful only if you have another system generating -unique IDs (for instance an SQL table) and you really want the Redis stream -IDs to match the one of this other system. +Redis guarantees that IDs are always incremental: the ID of any entry you insert will be greater than any previous ID, so entries are totally ordered inside a stream. To guarantee this property, if the current top ID in the stream has a time greater than the current local time of the instance, Redis uses the top entry time instead and increments the sequence part of the ID. This may happen when, for instance, the local clock jumps backward, or after a failover when the new master has a different absolute time. + +When you specify an explicit ID to `XADD`, the minimum valid ID is `0-1`, and you *must* specify an ID that is greater than any other ID currently inside the stream, otherwise the command fails and returns an error. Specifying explicit IDs is usually useful only if you have another system generating unique IDs (for instance an SQL table) and you want the Redis stream IDs to match those from your other system. ## Capped streams `XADD` incorporates the same semantics as the [`XTRIM`]({{< relref "/commands/xtrim" >}}) command - refer to its documentation page for more information. -This allows adding new entries and keeping the stream's size in check with a single call to `XADD`, effectively capping the stream with an arbitrary threshold. -Although exact trimming is possible and is the default, due to the internal representation of streams it is more efficient to add an entry and trim stream with `XADD` using **almost exact** trimming (the `~` argument). +This allows you to add new entries and keep the stream's size in check with a single call to `XADD`, effectively capping the stream with an arbitrary threshold. +Although exact trimming is possible and is the default, due to the internal representation of streams, it is more efficient to add an entry and trim the stream with `XADD` using **almost exact** trimming (the `~` argument). For example, calling `XADD` in the following form: XADD mystream MAXLEN ~ 1000 * ... entry fields here ... - -Will add a new entry but will also evict old entries so that the stream will contain only 1000 entries, or at most a few tens more. + +This adds a new entry but also evicts old entries so that the stream contains only 1000 entries, or at most a few tens more. ## Additional information about streams -For further information about Redis streams please check our +For more information about Redis streams, see the [introduction to Redis Streams document]({{< relref "/develop/data-types/streams" >}}). ## Examples diff --git a/content/commands/xdelex.md b/content/commands/xdelex.md new file mode 100644 index 0000000000..35a2858d78 --- /dev/null +++ b/content/commands/xdelex.md @@ -0,0 +1,149 @@ +--- +acl_categories: +- '@write' +- '@stream' +- '@fast' +arguments: +- display_text: key + key_spec_index: 0 + name: key + type: key +- arguments: + - display_text: keepref + name: keepref + token: KEEPREF + type: pure-token + - display_text: delref + name: delref + token: DELREF + type: pure-token + - display_text: acked + name: acked + token: ACKED + type: pure-token + name: condition + optional: true + type: oneof +- arguments: + - display_text: numids + name: numids + type: integer + - display_text: id + multiple: true + name: id + type: string + name: ids + token: IDS + type: block +arity: -5 +categories: +- docs +- develop +- stack +- oss +- rs +- rc +- oss +- kubernetes +- clients +command_flags: +- write +- fast +complexity: O(1) for each single item to delete in the stream, regardless of the stream + size. +description: Deletes one or multiple entries from the stream. +group: stream +hidden: false +key_specs: +- RW: true + begin_search: + spec: + index: 1 + type: index + delete: true + find_keys: + spec: + keystep: 1 + lastkey: 0 + limit: 0 + type: range +linkTitle: XDELEX +since: 8.2.0 +summary: Deletes one or multiple entries from the stream. +syntax_fmt: "XDELEX key [KEEPREF | DELREF | ACKED] IDS\_numids id [id ...]" +syntax_str: "[KEEPREF | DELREF | ACKED] IDS\_numids id [id ...]" +title: XDELEX +--- + +Deletes one or multiple entries from the stream at the specified `key`. + +`XDELEX` is an extension of the Redis Streams [`XDEL`]({{< relref "/commands/xdel" >}}) command that provides more control over how message entries are deleted concerning consumer groups. + +## Required arguments + +
+key + +The name of the stream key. +
+ +
+IDS numids id [id ...] + +The IDS block specifying which entries to delete: +- `numids`: The number of IDs that follow +- `id [id ...]`: One or more stream entry IDs to delete + +Note: The IDS block can be at any position in the command, same as other commands. +
+ +## Optional arguments + +
+KEEPREF | DELREF | ACKED + +Specifies how to handle consumer group references when deleting entries. Available since Redis 8.2. If no option is specified, `KEEPREF` is used by default: + +- `KEEPREF` (default): Deletes the specified entries from the stream, but preserves existing references to these entries in all consumer groups' PEL (Pending Entries List). This behavior is similar to [`XDEL`]({{< relref "/commands/xdel" >}}). +- `DELREF`: Deletes the specified entries from the stream and also removes all references to these entries from all consumer groups' pending entry lists, effectively cleaning up all traces of the messages. If an entry ID is not in the stream, but there are dangling references, `XDELEX` with `DELREF` would still remove all those references. +- `ACKED`: Only deletes entries that were read and acknowledged by all consumer groups. +
+ +The command provides fine-grained control over stream entry deletion, particularly useful when working with consumer groups where you need to manage pending entry references carefully. + +## Examples + +{{% redis-cli %}} +XADD mystream * field1 value1 +XADD mystream * field2 value2 +XADD mystream * field3 value3 +XRANGE mystream - + +XDELEX mystream KEEPREF IDS 2 1526919030474-55 1526919030474-56 +XRANGE mystream - + +{{% /redis-cli %}} + +## Return information + +{{< multitabs id="xdelex-return-info" + tab1="RESP2" + tab2="RESP3" >}} + +One of the following: + +* [Array reply]({{< relref "/develop/reference/protocol-spec#arrays" >}}): -1 for each requested ID when the given key does not exist. +* [Array reply](../../develop/reference/protocol-spec#arrays): For each ID: + * [Integer reply](../../develop/reference/protocol-spec#integers): -1 if no such ID exists in the provided stream key. + * [Integer reply](../../develop/reference/protocol-spec#integers): 1 if the entry was deleted from the stream. + * [Integer reply](../../develop/reference/protocol-spec#integers): 2 if the entry was not deleted, but there are still dangling references (ACKED option). + +-tab-sep- + +One of the following: + +* [Array reply]({{< relref "/develop/reference/protocol-spec#arrays" >}}): -1 for each requested ID when the given key does not exist. +* [Array reply](../../develop/reference/protocol-spec#arrays): For each ID: + * [Integer reply](../../develop/reference/protocol-spec#integers): -1 if no such ID exists in the provided stream key. + * [Integer reply](../../develop/reference/protocol-spec#integers): 1 if the entry was deleted from the stream. + * [Integer reply](../../develop/reference/protocol-spec#integers): 2 if the entry was not deleted, but there are still dangling references (ACKED option). + +{{< /multitabs >}} diff --git a/content/commands/xtrim.md b/content/commands/xtrim.md index 9086919bd9..fa40922f03 100644 --- a/content/commands/xtrim.md +++ b/content/commands/xtrim.md @@ -42,6 +42,22 @@ arguments: since: 6.2.0 token: LIMIT type: integer + - arguments: + - display_text: keepref + name: keepref + token: KEEPREF + type: pure-token + - display_text: delref + name: delref + token: DELREF + type: pure-token + - display_text: acked + name: acked + token: ACKED + type: pure-token + name: condition + optional: true + type: oneof name: trim type: block arity: -4 @@ -68,6 +84,8 @@ hints: history: - - 6.2.0 - Added the `MINID` trimming strategy and the `LIMIT` option. +- - 8.2.0 + - Added the `KEEPREF`, `DELREF` and `ACKED` options. key_specs: - RW: true begin_search: @@ -84,40 +102,91 @@ key_specs: linkTitle: XTRIM since: 5.0.0 summary: Deletes messages from the beginning of a stream. -syntax_fmt: "XTRIM key [= | ~] threshold [LIMIT\_count]" -syntax_str: " [= | ~] threshold [LIMIT\_count]" +syntax_fmt: "XTRIM key [= | ~] threshold [LIMIT\_count] [KEEPREF\n\ + \ | DELREF | ACKED]" +syntax_str: " [= | ~] threshold [LIMIT\_count] [KEEPREF | DELREF |\ + \ ACKED]" title: XTRIM --- + `XTRIM` trims the stream by evicting older entries (entries with lower IDs) if needed. -Trimming the stream can be done using one of these strategies: +## Required arguments + +
+key + +The name of the stream key. +
+ +
+MAXLEN | MINID + +The trimming strategy: +- `MAXLEN`: Evicts entries as long as the stream's length exceeds the specified threshold +- `MINID`: Evicts entries with IDs lower than the specified threshold (available since Redis 6.2.0) +
+ +
+threshold + +The trimming threshold. For `MAXLEN`, this is a positive integer representing the maximum number of entries. For `MINID`, this is a stream ID. +
+ +## Optional arguments + +
+= | ~ + +The trimming operator: +- `=`: Exact trimming (default) - trims to the exact threshold +- `~`: Approximate trimming - more efficient, may leave slightly more entries than the threshold +
+ +
+LIMIT count + +Limits the number of entries to examine during trimming. Available since Redis 6.2.0. When not specified, Redis uses a default value of 100 * the number of entries in a macro node. Specifying 0 disables the limiting mechanism entirely. +
+ +
+KEEPREF | DELREF | ACKED + +Specifies how to handle consumer group references when trimming. If no option is specified, `KEEPREF` is used by default: + +- `KEEPREF` (default): When trimming, removes entries from the stream according to the specified strategy (`MAXLEN` or `MINID`), regardless of whether they are referenced by any consumer groups, but preserves existing references to these entries in all consumer groups' PEL (Pending Entries List). +- `DELREF`: When trimming, removes entries from the stream according to the specified strategy and also removes all references to these entries from all consumer groups' PEL. +- `ACKED`: When trimming, only removes entries that were read and acknowledged by all consumer groups. Note that if the number of referenced entries is larger than `MAXLEN`, trimming will still stop at the limit. +
+ +You can trim the stream using one of these strategies: * `MAXLEN`: Evicts entries as long as the stream's length exceeds the specified `threshold`, where `threshold` is a positive integer. * `MINID`: Evicts entries with IDs lower than `threshold`, where `threshold` is a stream ID. -For example, this will trim the stream to exactly the latest 1000 items: +For example, this trims the stream to exactly the latest 1000 items: ``` XTRIM mystream MAXLEN 1000 ``` -Whereas in this example, all entries that have an ID lower than 649085820-0 will be evicted: +In this example, Redis evicts all entries that have an ID lower than 649085820-0: ``` XTRIM mystream MINID 649085820 ``` -By default, or when provided with the optional `=` argument, the command performs exact trimming. +By default, or when you provide the optional `=` argument, the command performs exact trimming. Depending on the strategy, exact trimming means: -* `MAXLEN`: the trimmed stream's length will be exactly the minimum between its original length and the specified `threshold`. -* `MINID`: the oldest ID in the stream will be exactly the maximum between its original oldest ID and the specified `threshold`. +* `MAXLEN`: The trimmed stream's length will be exactly the minimum between its original length and the specified `threshold`. +* `MINID`: The oldest ID in the stream will be exactly the maximum between its original oldest ID and the specified `threshold`. Nearly exact trimming --- -Because exact trimming may require additional effort from the Redis server, the optional `~` argument can be provided to make it more efficient. +Because exact trimming may require additional effort from the Redis server, you can provide the optional `~` argument to make it more efficient. For example: @@ -125,13 +194,13 @@ For example: XTRIM mystream MAXLEN ~ 1000 ``` -The `~` argument between the `MAXLEN` strategy and the `threshold` means that the user is requesting to trim the stream so its length is **at least** the `threshold`, but possibly slightly more. -In this case, Redis will stop trimming early when performance can be gained (for example, when a whole macro node in the data structure can't be removed). -This makes trimming much more efficient, and it is usually what you want, although after trimming, the stream may have few tens of additional entries over the `threshold`. +The `~` argument between the `MAXLEN` strategy and the `threshold` means that you are requesting to trim the stream so its length is **at least** the `threshold`, but possibly slightly more. +In this case, Redis stops trimming early when performance can be gained (for example, when a whole macro node in the data structure can't be removed). +This makes trimming much more efficient, and it is usually what you want, although after trimming, the stream may have a few tens of additional entries over the `threshold`. -Another way to control the amount of work done by the command when using the `~`, is the `LIMIT` clause. -When used, it specifies the maximal `count` of entries that will be evicted. -When `LIMIT` and `count` aren't specified, the default value of 100 * the number of entries in a macro node will be implicitly used as the `count`. +Another way to control the amount of work done by the command when using `~` is the `LIMIT` clause. +When you use it, it specifies the maximum `count` of entries that will be evicted. +When you don't specify `LIMIT` and `count`, Redis implicitly uses the default value of 100 * the number of entries in a macro node as the `count`. Specifying the value 0 as `count` disables the limiting mechanism entirely. ## Examples diff --git a/content/develop/data-types/streams.md b/content/develop/data-types/streams.md index 7f4b9dfd9f..3a1434401f 100644 --- a/content/develop/data-types/streams.md +++ b/content/develop/data-types/streams.md @@ -9,9 +9,7 @@ categories: - oss - kubernetes - clients -description: 'Introduction to Redis streams - - ' +description: Introduction to Redis streams linkTitle: Streams title: Redis Streams weight: 60 @@ -28,16 +26,18 @@ Examples of Redis stream use cases include: Redis generates a unique ID for each stream entry. You can use these IDs to retrieve their associated entries later or to read and process all subsequent entries in the stream. Note that because these IDs are related to time, the ones shown here may vary and will be different from the IDs you see in your own Redis instance. -Redis streams support several trimming strategies (to prevent streams from growing unbounded) and more than one consumption strategy (see [`XREAD`]({{< relref "/commands/xread" >}}), [`XREADGROUP`]({{< relref "/commands/xreadgroup" >}}), and [`XRANGE`]({{< relref "/commands/xrange" >}})). +Redis streams support several trimming strategies (to prevent streams from growing unbounded) and more than one consumption strategy (see [`XREAD`]({{< relref "/commands/xread" >}}), [`XREADGROUP`]({{< relref "/commands/xreadgroup" >}}), and [`XRANGE`]({{< relref "/commands/xrange" >}})). Starting with Redis 8.2, the `XACKDEL`, `XDELEX`, `XADD`, and `XTRIM` commands provide fine-grained control over how stream operations interact with multiple consumer groups, simplifying the coordination of message processing across different applications. ## Basic commands + * [`XADD`]({{< relref "/commands/xadd" >}}) adds a new entry to a stream. * [`XREAD`]({{< relref "/commands/xread" >}}) reads one or more entries, starting at a given position and moving forward in time. * [`XRANGE`]({{< relref "/commands/xrange" >}}) returns a range of entries between two supplied entry IDs. * [`XLEN`]({{< relref "/commands/xlen" >}}) returns the length of a stream. - -See the [complete list of stream commands]({{< relref "/commands/" >}}?group=stream). +* [`XDEL`]({{< relref "/commands/xdel" >}}) removes entries from a stream. +* [`XTRIM`]({{< relref "/commands/xtrim" >}}) trims a stream by removing older entries. +See the [complete list of stream commands]({{< relref "/commands/" >}}?group=stream). ## Examples @@ -90,7 +90,6 @@ For details on why, note that streams are implemented as [radix trees](https://e Simply put, Redis streams provide highly efficient inserts and reads. See each command's time complexity for the details. - ## Streams basics Streams are an append-only data structure. The fundamental write command, called [`XADD`]({{< relref "/commands/xadd" >}}), appends a new entry to the specified stream. @@ -393,6 +392,7 @@ Now it's time to zoom in to see the fundamental consumer group commands. They ar * [`XGROUP`]({{< relref "/commands/xgroup" >}}) is used in order to create, destroy and manage consumer groups. * [`XREADGROUP`]({{< relref "/commands/xreadgroup" >}}) is used to read from a stream via a consumer group. * [`XACK`]({{< relref "/commands/xack" >}}) is the command that allows a consumer to mark a pending message as correctly processed. +* [`XACKDEL`]({{< relref "/commands/xackdel" >}}) combines acknowledgment and deletion in a single atomic operation with enhanced control over consumer group references. ## Creating a consumer group @@ -681,6 +681,31 @@ The counter that you observe in the [`XPENDING`]({{< relref "/commands/xpending" When there are failures, it is normal that messages will be delivered multiple times, but eventually they usually get processed and acknowledged. However there might be a problem processing some specific message, because it is corrupted or crafted in a way that triggers a bug in the processing code. In such a case what happens is that consumers will continuously fail to process this particular message. Because we have the counter of the delivery attempts, we can use that counter to detect messages that for some reason are not processable. So once the deliveries counter reaches a given large number that you chose, it is probably wiser to put such messages in another stream and send a notification to the system administrator. This is basically the way that Redis Streams implements the *dead letter* concept. +## Working with multiple consumer groups + +Redis Streams can be associated with multiple consumer groups, where each entry is delivered to all the stream's consumer groups. Within each consumer group, consumers handle a portion of the entries collaboratively. This design enables different applications or services to process the same stream data independently. + +When a consumer processes a message, it acknowledges it using the [`XACK`]({{< relref "/commands/xack" >}}) command, which removes the entry reference from the Pending Entries List (PEL) of that specific consumer group. However, the entry remains in the stream and in the PELs of other consumer groups until they also acknowledge it. + +Traditionally, applications needed to implement complex logic to delete entries from the stream only after all consumer groups had acknowledged them. This coordination was challenging to implement correctly and efficiently. + +### Enhanced deletion control in Redis 8.2 + +Starting with Redis 8.2, several commands provide enhanced control over how entries are handled with respect to multiple consumer groups: + +* [`XADD`]({{< relref "/commands/xadd" >}}) with trimming options now supports `KEEPREF`, `DELREF`, and `ACKED` modes +* [`XTRIM`]({{< relref "/commands/xtrim" >}}) supports the same reference handling options +* [`XDELEX`]({{< relref "/commands/xdelex" >}}) provides fine-grained deletion control +* [`XACKDEL`]({{< relref "/commands/xackdel" >}}) combines acknowledgment and deletion atomically + +These options control how consumer group references are handled: + +- **KEEPREF** (default): Preserves existing references to entries in all consumer groups' PELs, maintaining backward compatibility +- **DELREF**: Removes all references to entries from all consumer groups' PELs, effectively cleaning up all traces of the messages +- **ACKED**: Only processes entries that have been acknowledged by all consumer groups + +The `ACKED` option is particularly useful as it automates the complex logic of coordinating deletion across multiple consumer groups, ensuring entries are only removed when all groups have finished processing them. + ## Streams observability Messaging systems that lack observability are very hard to work with. Not knowing who is consuming messages, what messages are pending, the set of consumer groups active in a given stream, makes everything opaque. For this reason, Redis Streams and consumer groups have different ways to observe what is happening. We already covered [`XPENDING`]({{< relref "/commands/xpending" >}}), which allows us to inspect the list of messages that are under processing at a given moment, together with their idle time and number of deliveries. @@ -832,6 +857,21 @@ However, [`XTRIM`]({{< relref "/commands/xtrim" >}}) is designed to accept diffe As [`XTRIM`]({{< relref "/commands/xtrim" >}}) is an explicit command, the user is expected to know about the possible shortcomings of different trimming strategies. +### Trimming with consumer group awareness + +Starting with Redis 8.2, both [`XADD`]({{< relref "/commands/xadd" >}}) with trimming options and [`XTRIM`]({{< relref "/commands/xtrim" >}}) support enhanced control over how trimming interacts with consumer groups through the `KEEPREF`, `DELREF`, and `ACKED` options: + +``` +XADD mystream KEEPREF MAXLEN 1000 * field value +XTRIM mystream ACKED MAXLEN 1000 +``` + +- **KEEPREF** (default): Trims entries according to the strategy but preserves references in consumer group PELs +- **DELREF**: Trims entries and removes all references from consumer group PELs +- **ACKED**: Only trims entries that have been acknowledged by all consumer groups + +The `ACKED` option is particularly useful for maintaining data integrity across multiple consumer groups, ensuring that entries are only removed when all groups have finished processing them. + Another useful eviction strategy that may be added to [`XTRIM`]({{< relref "/commands/xtrim" >}}) in the future, is to remove by a range of IDs to ease use of [`XRANGE`]({{< relref "/commands/xrange" >}}) and [`XTRIM`]({{< relref "/commands/xtrim" >}}) to move data from Redis to other storage systems if needed. ## Special IDs in the streams API @@ -882,7 +922,15 @@ Streams also have a special command for removing items from the middle of a stre 2) "Wood" {{< /clients-example >}} -However in the current implementation, memory is not really reclaimed until a macro node is completely empty, so you should not abuse this feature. +### Enhanced deletion with XDELEX + +Starting with Redis 8.2, the [`XDELEX`]({{< relref "/commands/xdelex" >}}) command provides enhanced control over entry deletion, particularly when working with consumer groups. Like other enhanced commands, it supports `KEEPREF`, `DELREF`, and `ACKED` options: + +``` +XDELEX mystream ACKED IDS 2 1692633198206-0 1692633208557-0 +``` + +This allows you to delete entries only when they have been acknowledged by all consumer groups (`ACKED`), remove all consumer group references (`DELREF`), or preserve existing references (`KEEPREF`). ## Zero length streams @@ -931,9 +979,6 @@ A few remarks: * Here we processed up to 10k messages per iteration, this means that the `COUNT` parameter of [`XREADGROUP`]({{< relref "/commands/xreadgroup" >}}) was set to 10000. This adds a lot of latency but is needed in order to allow the slow consumers to be able to keep with the message flow. So you can expect a real world latency that is a lot smaller. * The system used for this benchmark is very slow compared to today's standards. - - - ## Learn more * The [Redis Streams Tutorial]({{< relref "/develop/data-types/streams" >}}) explains Redis streams with many examples.