Skip to content

Commit 2dab656

Browse files
mgravellmgravell
authored
propose XNACK for 8.8 (#3058)
* propose XNACK for 8.8 * PR number
1 parent 6e5b950 commit 2dab656

14 files changed

Lines changed: 307 additions & 2 deletions

File tree

‎docs/ReleaseNotes.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ Current package versions:
99
## Unreleased
1010

1111
- Detect server-mode correctly on Valkey 8+ instances ([#3050 by @wipiano](https://github.com/StackExchange/StackExchange.Redis/pull/3050))
12+
- Add Redis 8.8 stream negative acknowledgements (`XNACK`) ([#3058 by @mgravell](https://github.com/StackExchange/StackExchange.Redis/pull/3058))
1213
- Update experimental `GCRA` APIs and wire protocol terminology from "requests" to "tokens", to match server change ([#3051 by @mgravell](https://github.com/StackExchange/StackExchange.Redis/pull/3051))
1314

1415
## 2.12.14

‎src/StackExchange.Redis/Enums/RedisCommand.cs‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -240,6 +240,7 @@ internal enum RedisCommand
240240
XGROUP,
241241
XINFO,
242242
XLEN,
243+
XNACK,
243244
XPENDING,
244245
XRANGE,
245246
XREAD,
@@ -561,6 +562,7 @@ internal static bool IsPrimaryOnly(this RedisCommand command)
561562
case RedisCommand.XDEL:
562563
case RedisCommand.XDELEX:
563564
case RedisCommand.XGROUP:
565+
case RedisCommand.XNACK:
564566
case RedisCommand.XREADGROUP:
565567
case RedisCommand.XTRIM:
566568
return false;

‎src/StackExchange.Redis/Enums/StreamNackMode.cs‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,26 @@
1+
using System.Diagnostics.CodeAnalysis;
2+
using RESPite;
3+
4+
namespace StackExchange.Redis;
5+
6+
/// <summary>
7+
/// Determines how a stream message is negatively acknowledged back to the consumer group.
8+
/// </summary>
9+
[Experimental(Experiments.Server_8_8, UrlFormat = Experiments.UrlFormat)]
10+
public enum StreamNackMode
11+
{
12+
/// <summary>
13+
/// Release the message without counting it as an additional failure.
14+
/// </summary>
15+
Silent = 0,
16+
17+
/// <summary>
18+
/// Release the message and treat it as a normal failed delivery.
19+
/// </summary>
20+
Fail = 1,
21+
22+
/// <summary>
23+
/// Release the message and mark it as a terminal failure.
24+
/// </summary>
25+
Fatal = 2,
26+
}

‎src/StackExchange.Redis/Interfaces/IDatabase.cs‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2614,6 +2614,36 @@ IEnumerable<SortedSetEntry> SortedSetScan(
26142614
StreamTrimResult[] StreamAcknowledgeAndDelete(RedisKey key, RedisValue groupName, StreamTrimMode mode, RedisValue[] messageIds, CommandFlags flags = CommandFlags.None);
26152615
#pragma warning restore RS0026
26162616

2617+
/// <summary>
2618+
/// Allow the consumer to release a pending message back to the group without marking it as correctly processed.
2619+
/// Returns the number of messages negatively acknowledged.
2620+
/// </summary>
2621+
/// <param name="key">The key of the stream.</param>
2622+
/// <param name="groupName">The name of the consumer group that received the message.</param>
2623+
/// <param name="consumerName">The name of the consumer releasing the message.</param>
2624+
/// <param name="mode">The negative acknowledge mode to use.</param>
2625+
/// <param name="messageId">The ID of the message to negatively acknowledge.</param>
2626+
/// <param name="flags">The flags to use for this operation.</param>
2627+
/// <returns>The number of messages negatively acknowledged.</returns>
2628+
/// <remarks><seealso href="https://redis.io/topics/streams-intro"/></remarks>
2629+
[Experimental(Experiments.Server_8_8, UrlFormat = Experiments.UrlFormat)]
2630+
long StreamNegativeAcknowledge(RedisKey key, RedisValue groupName, RedisValue consumerName, StreamNackMode mode, RedisValue messageId, CommandFlags flags = CommandFlags.None);
2631+
2632+
/// <summary>
2633+
/// Allow the consumer to release pending messages back to the group without marking them as correctly processed.
2634+
/// Returns the number of messages negatively acknowledged.
2635+
/// </summary>
2636+
/// <param name="key">The key of the stream.</param>
2637+
/// <param name="groupName">The name of the consumer group that received the messages.</param>
2638+
/// <param name="consumerName">The name of the consumer releasing the messages.</param>
2639+
/// <param name="mode">The negative acknowledge mode to use.</param>
2640+
/// <param name="messageIds">The IDs of the messages to negatively acknowledge.</param>
2641+
/// <param name="flags">The flags to use for this operation.</param>
2642+
/// <returns>The number of messages negatively acknowledged.</returns>
2643+
/// <remarks><seealso href="https://redis.io/topics/streams-intro"/></remarks>
2644+
[Experimental(Experiments.Server_8_8, UrlFormat = Experiments.UrlFormat)]
2645+
long StreamNegativeAcknowledge(RedisKey key, RedisValue groupName, RedisValue consumerName, StreamNackMode mode, RedisValue[] messageIds, CommandFlags flags = CommandFlags.None);
2646+
26172647
/// <summary>
26182648
/// Adds an entry using the specified values to the given stream key.
26192649
/// If key does not exist, a new key holding a stream is created.

‎src/StackExchange.Redis/Interfaces/IDatabaseAsync.cs‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -644,6 +644,14 @@ IAsyncEnumerable<SortedSetEntry> SortedSetScanAsync(
644644
Task<StreamTrimResult[]> StreamAcknowledgeAndDeleteAsync(RedisKey key, RedisValue groupName, StreamTrimMode mode, RedisValue[] messageIds, CommandFlags flags = CommandFlags.None);
645645
#pragma warning restore RS0026
646646

647+
/// <inheritdoc cref="IDatabase.StreamNegativeAcknowledge(RedisKey, RedisValue, RedisValue, StreamNackMode, RedisValue, CommandFlags)"/>
648+
[Experimental(Experiments.Server_8_8, UrlFormat = Experiments.UrlFormat)]
649+
Task<long> StreamNegativeAcknowledgeAsync(RedisKey key, RedisValue groupName, RedisValue consumerName, StreamNackMode mode, RedisValue messageId, CommandFlags flags = CommandFlags.None);
650+
651+
/// <inheritdoc cref="IDatabase.StreamNegativeAcknowledge(RedisKey, RedisValue, RedisValue, StreamNackMode, RedisValue[], CommandFlags)"/>
652+
[Experimental(Experiments.Server_8_8, UrlFormat = Experiments.UrlFormat)]
653+
Task<long> StreamNegativeAcknowledgeAsync(RedisKey key, RedisValue groupName, RedisValue consumerName, StreamNackMode mode, RedisValue[] messageIds, CommandFlags flags = CommandFlags.None);
654+
647655
/// <inheritdoc cref="IDatabase.StreamAdd(RedisKey, RedisValue, RedisValue, RedisValue?, int?, bool, CommandFlags)"/>
648656
Task<RedisValue> StreamAddAsync(RedisKey key, RedisValue streamField, RedisValue streamValue, RedisValue? messageId, int? maxLength, bool useApproximateMaxLength, CommandFlags flags);
649657

‎src/StackExchange.Redis/KeyspaceIsolation/KeyPrefixed.cs‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -609,6 +609,12 @@ public Task<StreamTrimResult> StreamAcknowledgeAndDeleteAsync(RedisKey key, Redi
609609
public Task<StreamTrimResult[]> StreamAcknowledgeAndDeleteAsync(RedisKey key, RedisValue groupName, StreamTrimMode mode, RedisValue[] messageIds, CommandFlags flags = CommandFlags.None) =>
610610
Inner.StreamAcknowledgeAndDeleteAsync(ToInner(key), groupName, mode, messageIds, flags);
611611

612+
public Task<long> StreamNegativeAcknowledgeAsync(RedisKey key, RedisValue groupName, RedisValue consumerName, StreamNackMode mode, RedisValue messageId, CommandFlags flags = CommandFlags.None) =>
613+
Inner.StreamNegativeAcknowledgeAsync(ToInner(key), groupName, consumerName, mode, messageId, flags);
614+
615+
public Task<long> StreamNegativeAcknowledgeAsync(RedisKey key, RedisValue groupName, RedisValue consumerName, StreamNackMode mode, RedisValue[] messageIds, CommandFlags flags = CommandFlags.None) =>
616+
Inner.StreamNegativeAcknowledgeAsync(ToInner(key), groupName, consumerName, mode, messageIds, flags);
617+
612618
public Task<RedisValue> StreamAddAsync(RedisKey key, RedisValue streamField, RedisValue streamValue, RedisValue? messageId, int? maxLength, bool useApproximateMaxLength, CommandFlags flags) =>
613619
Inner.StreamAddAsync(ToInner(key), streamField, streamValue, messageId, maxLength, useApproximateMaxLength, flags);
614620

‎src/StackExchange.Redis/KeyspaceIsolation/KeyPrefixedDatabase.cs‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -591,6 +591,12 @@ public StreamTrimResult StreamAcknowledgeAndDelete(RedisKey key, RedisValue grou
591591
public StreamTrimResult[] StreamAcknowledgeAndDelete(RedisKey key, RedisValue groupName, StreamTrimMode mode, RedisValue[] messageIds, CommandFlags flags = CommandFlags.None) =>
592592
Inner.StreamAcknowledgeAndDelete(ToInner(key), groupName, mode, messageIds, flags);
593593

594+
public long StreamNegativeAcknowledge(RedisKey key, RedisValue groupName, RedisValue consumerName, StreamNackMode mode, RedisValue messageId, CommandFlags flags = CommandFlags.None) =>
595+
Inner.StreamNegativeAcknowledge(ToInner(key), groupName, consumerName, mode, messageId, flags);
596+
597+
public long StreamNegativeAcknowledge(RedisKey key, RedisValue groupName, RedisValue consumerName, StreamNackMode mode, RedisValue[] messageIds, CommandFlags flags = CommandFlags.None) =>
598+
Inner.StreamNegativeAcknowledge(ToInner(key), groupName, consumerName, mode, messageIds, flags);
599+
594600
public RedisValue StreamAdd(RedisKey key, RedisValue streamField, RedisValue streamValue, RedisValue? messageId, int? maxLength, bool useApproximateMaxLength, CommandFlags flags) =>
595601
Inner.StreamAdd(ToInner(key), streamField, streamValue, messageId, maxLength, useApproximateMaxLength, flags);
596602

‎src/StackExchange.Redis/PublicAPI/PublicAPI.Shipped.txt‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2258,9 +2258,13 @@ static StackExchange.Redis.RedisChannel.KeySpaceSingleKey(in StackExchange.Redis
22582258
[SER003]StackExchange.Redis.IDatabase.StreamAdd(StackExchange.Redis.RedisKey key, StackExchange.Redis.NameValueEntry[]! streamPairs, StackExchange.Redis.StreamIdempotentId idempotentId, long? maxLength = null, bool useApproximateMaxLength = false, long? limit = null, StackExchange.Redis.StreamTrimMode trimMode = StackExchange.Redis.StreamTrimMode.KeepReferences, StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> StackExchange.Redis.RedisValue
22592259
[SER003]StackExchange.Redis.IDatabase.StreamAdd(StackExchange.Redis.RedisKey key, StackExchange.Redis.RedisValue streamField, StackExchange.Redis.RedisValue streamValue, StackExchange.Redis.StreamIdempotentId idempotentId, long? maxLength = null, bool useApproximateMaxLength = false, long? limit = null, StackExchange.Redis.StreamTrimMode trimMode = StackExchange.Redis.StreamTrimMode.KeepReferences, StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> StackExchange.Redis.RedisValue
22602260
[SER003]StackExchange.Redis.IDatabase.StreamConfigure(StackExchange.Redis.RedisKey key, StackExchange.Redis.StreamConfiguration! configuration, StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> void
2261+
[SER006]StackExchange.Redis.IDatabase.StreamNegativeAcknowledge(StackExchange.Redis.RedisKey key, StackExchange.Redis.RedisValue groupName, StackExchange.Redis.RedisValue consumerName, StackExchange.Redis.StreamNackMode mode, StackExchange.Redis.RedisValue messageId, StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> long
2262+
[SER006]StackExchange.Redis.IDatabase.StreamNegativeAcknowledge(StackExchange.Redis.RedisKey key, StackExchange.Redis.RedisValue groupName, StackExchange.Redis.RedisValue consumerName, StackExchange.Redis.StreamNackMode mode, StackExchange.Redis.RedisValue[]! messageIds, StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> long
22612263
[SER003]StackExchange.Redis.IDatabaseAsync.StreamAddAsync(StackExchange.Redis.RedisKey key, StackExchange.Redis.NameValueEntry[]! streamPairs, StackExchange.Redis.StreamIdempotentId idempotentId, long? maxLength = null, bool useApproximateMaxLength = false, long? limit = null, StackExchange.Redis.StreamTrimMode trimMode = StackExchange.Redis.StreamTrimMode.KeepReferences, StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> System.Threading.Tasks.Task<StackExchange.Redis.RedisValue>!
22622264
[SER003]StackExchange.Redis.IDatabaseAsync.StreamAddAsync(StackExchange.Redis.RedisKey key, StackExchange.Redis.RedisValue streamField, StackExchange.Redis.RedisValue streamValue, StackExchange.Redis.StreamIdempotentId idempotentId, long? maxLength = null, bool useApproximateMaxLength = false, long? limit = null, StackExchange.Redis.StreamTrimMode trimMode = StackExchange.Redis.StreamTrimMode.KeepReferences, StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> System.Threading.Tasks.Task<StackExchange.Redis.RedisValue>!
22632265
[SER003]StackExchange.Redis.IDatabaseAsync.StreamConfigureAsync(StackExchange.Redis.RedisKey key, StackExchange.Redis.StreamConfiguration! configuration, StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> System.Threading.Tasks.Task!
2266+
[SER006]StackExchange.Redis.IDatabaseAsync.StreamNegativeAcknowledgeAsync(StackExchange.Redis.RedisKey key, StackExchange.Redis.RedisValue groupName, StackExchange.Redis.RedisValue consumerName, StackExchange.Redis.StreamNackMode mode, StackExchange.Redis.RedisValue messageId, StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> System.Threading.Tasks.Task<long>!
2267+
[SER006]StackExchange.Redis.IDatabaseAsync.StreamNegativeAcknowledgeAsync(StackExchange.Redis.RedisKey key, StackExchange.Redis.RedisValue groupName, StackExchange.Redis.RedisValue consumerName, StackExchange.Redis.StreamNackMode mode, StackExchange.Redis.RedisValue[]! messageIds, StackExchange.Redis.CommandFlags flags = StackExchange.Redis.CommandFlags.None) -> System.Threading.Tasks.Task<long>!
22642268
[SER003]StackExchange.Redis.StreamConfiguration
22652269
[SER003]StackExchange.Redis.StreamConfiguration.IdmpDuration.get -> long?
22662270
[SER003]StackExchange.Redis.StreamConfiguration.IdmpDuration.set -> void
@@ -2279,6 +2283,10 @@ static StackExchange.Redis.RedisChannel.KeySpaceSingleKey(in StackExchange.Redis
22792283
[SER003]StackExchange.Redis.StreamInfo.IidsDuplicates.get -> long
22802284
[SER003]StackExchange.Redis.StreamInfo.IidsTracked.get -> long
22812285
[SER003]StackExchange.Redis.StreamInfo.PidsTracked.get -> long
2286+
[SER006]StackExchange.Redis.StreamNackMode
2287+
[SER006]StackExchange.Redis.StreamNackMode.Fail = 1 -> StackExchange.Redis.StreamNackMode
2288+
[SER006]StackExchange.Redis.StreamNackMode.Fatal = 2 -> StackExchange.Redis.StreamNackMode
2289+
[SER006]StackExchange.Redis.StreamNackMode.Silent = 0 -> StackExchange.Redis.StreamNackMode
22822290
StackExchange.Redis.StreamInfo.EntriesAdded.get -> long
22832291
StackExchange.Redis.StreamInfo.MaxDeletedEntryId.get -> StackExchange.Redis.RedisValue
22842292
StackExchange.Redis.StreamInfo.RecordedFirstEntryId.get -> StackExchange.Redis.RedisValue

‎src/StackExchange.Redis/RedisFeatures.cs‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -49,7 +49,8 @@ namespace StackExchange.Redis
4949
v8_0_0_M04 = new Version(7, 9, 227), // 8.0 M04 is version 7.9.227
5050
v8_2_0_rc1 = new Version(8, 1, 240), // 8.2 RC1 is version 8.1.240
5151
v8_4_0_rc1 = new Version(8, 3, 224), // 8.4 RC1 is version 8.3.224
52-
v8_6_0 = new Version(8, 6, 0);
52+
v8_6_0 = new Version(8, 6, 0),
53+
v8_8_0 = new Version(8, 8, 0);
5354

5455
#pragma warning restore SA1310 // Field names should not contain underscore
5556
#pragma warning restore SA1311 // Static readonly fields should begin with upper-case letter

0 commit comments

Comments
 (0)