diff --git a/src/NATS.Client.JetStream/INatsJSContext.cs b/src/NATS.Client.JetStream/INatsJSContext.cs index 7b5b62c7b..bc98adb4b 100644 --- a/src/NATS.Client.JetStream/INatsJSContext.cs +++ b/src/NATS.Client.JetStream/INatsJSContext.cs @@ -195,6 +195,21 @@ IAsyncEnumerable ListConsumerNamesAsync( /// The name is null. ValueTask UnpinConsumerAsync(string stream, string consumer, string group, CancellationToken cancellationToken = default); + /// + /// Reset a consumer's delivery state. + /// + /// Stream name where consumer is associated to. + /// Consumer name to be reset. + /// Stream sequence to reset to. Zero (the default) resets the consumer to its current ack floor. + /// A used to cancel the API call. + /// The reset response, including the consumer info and the sequence the consumer was reset to. + /// There was an issue retrieving the response. + /// Server responded with an error. + /// The name is invalid. + /// The name is null. + /// This feature is only available on NATS server v2.14 and later. + ValueTask ResetConsumerAsync(string stream, string consumer, ulong seq = 0, CancellationToken cancellationToken = default); + /// /// Calls JetStream Account Info API. /// diff --git a/src/NATS.Client.JetStream/Models/ConsumerResetRequest.cs b/src/NATS.Client.JetStream/Models/ConsumerResetRequest.cs new file mode 100644 index 000000000..206361488 --- /dev/null +++ b/src/NATS.Client.JetStream/Models/ConsumerResetRequest.cs @@ -0,0 +1,17 @@ +namespace NATS.Client.JetStream.Models; + +/// +/// A request to the JetStream $JS.API.CONSUMER.RESET API +/// +/// This feature is only available on NATS server v2.14 and later. +internal record ConsumerResetRequest +{ + /// + /// Stream sequence to reset the consumer to. Zero (the default) is sent as an empty body + /// and resets the consumer to its current ack floor. + /// + [System.Text.Json.Serialization.JsonPropertyName("seq")] + [System.Text.Json.Serialization.JsonIgnore(Condition = System.Text.Json.Serialization.JsonIgnoreCondition.WhenWritingDefault)] + [System.ComponentModel.DataAnnotations.Range(ulong.MinValue, ulong.MaxValue)] + public ulong Seq { get; set; } +} diff --git a/src/NATS.Client.JetStream/Models/ConsumerResetResponse.cs b/src/NATS.Client.JetStream/Models/ConsumerResetResponse.cs new file mode 100644 index 000000000..9aaf7e758 --- /dev/null +++ b/src/NATS.Client.JetStream/Models/ConsumerResetResponse.cs @@ -0,0 +1,16 @@ +namespace NATS.Client.JetStream.Models; + +/// +/// A response from the JetStream $JS.API.CONSUMER.RESET API +/// + +public record ConsumerResetResponse : ConsumerInfo +{ + /// + /// The stream sequence the consumer was reset to. The next delivered message will have a stream sequence >= ResetSeq. + /// + [System.Text.Json.Serialization.JsonPropertyName("reset_seq")] + [System.Text.Json.Serialization.JsonIgnore(Condition = System.Text.Json.Serialization.JsonIgnoreCondition.Never)] + [System.ComponentModel.DataAnnotations.Range(ulong.MinValue, ulong.MaxValue)] + public ulong ResetSeq { get; set; } +} diff --git a/src/NATS.Client.JetStream/NatsJSContext.Consumers.cs b/src/NATS.Client.JetStream/NatsJSContext.Consumers.cs index fabc576e3..9e9dc071d 100644 --- a/src/NATS.Client.JetStream/NatsJSContext.Consumers.cs +++ b/src/NATS.Client.JetStream/NatsJSContext.Consumers.cs @@ -188,6 +188,16 @@ await JSRequestResponseAsync( cancellationToken); } + /// + public async ValueTask ResetConsumerAsync(string stream, string consumer, ulong seq = 0, CancellationToken cancellationToken = default) + { + ThrowIfInvalidStreamName(stream); + return await JSRequestResponseAsync( + subject: $"{Opts.Prefix}.CONSUMER.RESET.{stream}.{consumer}", + request: seq == 0 ? null : new ConsumerResetRequest { Seq = seq }, + cancellationToken); + } + internal ValueTask CreateOrderedConsumerInternalAsync( string stream, NatsJSOrderedConsumerOpts opts, diff --git a/src/NATS.Client.JetStream/NatsJSJsonSerializer.cs b/src/NATS.Client.JetStream/NatsJSJsonSerializer.cs index d581d7aa7..7b36ee817 100644 --- a/src/NATS.Client.JetStream/NatsJSJsonSerializer.cs +++ b/src/NATS.Client.JetStream/NatsJSJsonSerializer.cs @@ -32,6 +32,8 @@ public static class NatsJSJsonSerializer [JsonSerializable(typeof(ConsumerNamesResponse))] [JsonSerializable(typeof(ConsumerPauseRequest))] [JsonSerializable(typeof(ConsumerPauseResponse))] +[JsonSerializable(typeof(ConsumerResetRequest))] +[JsonSerializable(typeof(ConsumerResetResponse))] [JsonSerializable(typeof(ConsumerUnpinRequest))] [JsonSerializable(typeof(ConsumerUnpinResponse))] [JsonSerializable(typeof(ErrorResponse))] diff --git a/tests/NATS.Client.JetStream.Tests/ManageConsumerTest.cs b/tests/NATS.Client.JetStream.Tests/ManageConsumerTest.cs index 19713d0a0..5be609300 100644 --- a/tests/NATS.Client.JetStream.Tests/ManageConsumerTest.cs +++ b/tests/NATS.Client.JetStream.Tests/ManageConsumerTest.cs @@ -135,6 +135,57 @@ public async Task Pause_resume_consumer() } } + [SkipIfNatsServer(versionEarlierThan: "2.14")] + public async Task Reset_consumer() + { + await using var nats = new NatsConnection(new NatsOpts { Url = _server.Url }); + await nats.ConnectRetryAsync(); + var prefix = _server.GetNextId(); + var js = new NatsJSContext(nats); + + var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); + + await js.CreateStreamAsync(new StreamConfig($"{prefix}s1", [$"{prefix}s1.*"]), cts.Token); + + for (var i = 1; i <= 5; i++) + { + var ack = await js.PublishAsync($"{prefix}s1.x", i, cancellationToken: cts.Token); + ack.EnsureSuccess(); + } + + var consumer = (NatsJSConsumer)await js.CreateOrUpdateConsumerAsync($"{prefix}s1", new ConsumerConfig($"{prefix}c1"), cts.Token); + + // DoubleAck so the server's ack floor is observably updated before we issue the reset. + var fetchOpts = new NatsJSFetchOpts { MaxMsgs = 2, Expires = TimeSpan.FromSeconds(5) }; + await foreach (var msg in consumer.FetchAsync(opts: fetchOpts, cancellationToken: cts.Token)) + { + await msg.AckAsync(new AckOpts { DoubleAck = true }, cancellationToken: cts.Token); + } + + // Reset to a specific stream sequence: next delivery should be that sequence. + { + var resetResponse = await js.ResetConsumerAsync($"{prefix}s1", $"{prefix}c1", seq: 4, cts.Token); + Assert.Equal(4ul, resetResponse.ResetSeq); + Assert.Equal($"{prefix}c1", resetResponse.Name); + + var next = await consumer.NextAsync(cancellationToken: cts.Token); + Assert.NotNull(next); + Assert.Equal(4ul, next!.Metadata!.Value.Sequence.Stream); + Assert.Equal(4, next.Data); + await next.AckAsync(new AckOpts { DoubleAck = true }, cancellationToken: cts.Token); + } + + // Reset with seq=0 (empty body): rewinds to the ack floor. ResetSeq is the next deliverable sequence. + { + var resetResponse = await js.ResetConsumerAsync($"{prefix}s1", $"{prefix}c1", cancellationToken: cts.Token); + Assert.Equal(5ul, resetResponse.ResetSeq); + + var next = await consumer.NextAsync(cancellationToken: cts.Token); + Assert.NotNull(next); + Assert.Equal(5ul, next!.Metadata!.Value.Sequence.Stream); + } + } + [SkipIfNatsServer(versionEarlierThan: "2.10")] public async Task Consumer_create_update_action() { diff --git a/tests/NATS.Client.KeyValueStore.Tests/NatsKVContextFactoryTest.cs b/tests/NATS.Client.KeyValueStore.Tests/NatsKVContextFactoryTest.cs index 6a769ee4a..65fe6f7bf 100644 --- a/tests/NATS.Client.KeyValueStore.Tests/NatsKVContextFactoryTest.cs +++ b/tests/NATS.Client.KeyValueStore.Tests/NatsKVContextFactoryTest.cs @@ -69,6 +69,8 @@ public class MockJsContext : INatsJSContext public ValueTask UnpinConsumerAsync(string stream, string consumer, string group, CancellationToken cancellationToken = default) => throw new NotImplementedException(); + public ValueTask ResetConsumerAsync(string stream, string consumer, ulong seq = 0, CancellationToken cancellationToken = default) => throw new NotImplementedException(); + public ValueTask GetAccountInfoAsync(CancellationToken cancellationToken = default) => throw new NotImplementedException(); public ValueTask PublishAsync(string subject, T? data, INatsSerialize? serializer = default, NatsJSPubOpts? opts = default, NatsHeaders? headers = default, CancellationToken cancellationToken = default) => throw new NotImplementedException(); diff --git a/tests/NATS.Client.ObjectStore.Tests/NatsObjContextFactoryTest.cs b/tests/NATS.Client.ObjectStore.Tests/NatsObjContextFactoryTest.cs index b9e675f61..58698f06a 100644 --- a/tests/NATS.Client.ObjectStore.Tests/NatsObjContextFactoryTest.cs +++ b/tests/NATS.Client.ObjectStore.Tests/NatsObjContextFactoryTest.cs @@ -69,6 +69,8 @@ public class MockJsContext : INatsJSContext public ValueTask UnpinConsumerAsync(string stream, string consumer, string group, CancellationToken cancellationToken = default) => throw new NotImplementedException(); + public ValueTask ResetConsumerAsync(string stream, string consumer, ulong seq = 0, CancellationToken cancellationToken = default) => throw new NotImplementedException(); + public ValueTask GetAccountInfoAsync(CancellationToken cancellationToken = default) => throw new NotImplementedException(); public ValueTask PublishAsync(string subject, T? data, INatsSerialize? serializer = default, NatsJSPubOpts? opts = default, NatsHeaders? headers = default, CancellationToken cancellationToken = default) => throw new NotImplementedException();