Add consumer reset API

src/NATS.Client.JetStream/INatsJSContext.cs

@@ -195,6 +195,21 @@ IAsyncEnumerable<string> ListConsumerNamesAsync(
     /// <exception cref="ArgumentNullException">The <paramref name="stream"/> name is <c>null</c>.</exception>
     ValueTask UnpinConsumerAsync(string stream, string consumer, string group, CancellationToken cancellationToken = default);
 
+    /// <summary>
+    /// Reset a consumer's delivery state.
+    /// </summary>
+    /// <param name="stream">Stream name where consumer is associated to.</param>
+    /// <param name="consumer">Consumer name to be reset.</param>
+    /// <param name="seq">Stream sequence to reset to. Zero (the default) resets the consumer to its current ack floor.</param>
+    /// <param name="cancellationToken">A <see cref="CancellationToken"/> used to cancel the API call.</param>
+    /// <returns>The reset response, including the consumer info and the sequence the consumer was reset to.</returns>
+    /// <exception cref="NatsJSException">There was an issue retrieving the response.</exception>
+    /// <exception cref="NatsJSApiException">Server responded with an error.</exception>
+    /// <exception cref="ArgumentException">The <paramref name="stream"/> name is invalid.</exception>
+    /// <exception cref="ArgumentNullException">The <paramref name="stream"/> name is <c>null</c>.</exception>
+    /// <remarks>This feature is only available on NATS server v2.14 and later.</remarks>
+    ValueTask<ConsumerResetResponse> ResetConsumerAsync(string stream, string consumer, ulong seq = 0, CancellationToken cancellationToken = default);
+
     /// <summary>
     /// Calls JetStream Account Info API.
     /// </summary>

src/NATS.Client.JetStream/Models/ConsumerResetRequest.cs

@@ -0,0 +1,17 @@
+namespace NATS.Client.JetStream.Models;
+
+/// <summary>
+/// A request to the JetStream $JS.API.CONSUMER.RESET API
+/// </summary>
+/// <remarks>This feature is only available on NATS server v2.14 and later.</remarks>
+internal record ConsumerResetRequest
+{
+    /// <summary>
+    /// 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.
+    /// </summary>
+    [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; }
+}

src/NATS.Client.JetStream/Models/ConsumerResetResponse.cs

@@ -0,0 +1,16 @@
+namespace NATS.Client.JetStream.Models;
+
+/// <summary>
+/// A response from the JetStream $JS.API.CONSUMER.RESET API
+/// </summary>
+
+public record ConsumerResetResponse : ConsumerInfo
+{
+    /// <summary>
+    /// The stream sequence the consumer was reset to. The next delivered message will have a stream sequence >= ResetSeq.
+    /// </summary>
+    [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; }
+}

src/NATS.Client.JetStream/NatsJSContext.Consumers.cs

@@ -188,6 +188,16 @@ await JSRequestResponseAsync<ConsumerUnpinRequest, ConsumerUnpinResponse>(
             cancellationToken);
     }
 
+    /// <inheritdoc />
+    public async ValueTask<ConsumerResetResponse> ResetConsumerAsync(string stream, string consumer, ulong seq = 0, CancellationToken cancellationToken = default)
+    {
+        ThrowIfInvalidStreamName(stream);
+        return await JSRequestResponseAsync<ConsumerResetRequest, ConsumerResetResponse>(
+            subject: $"{Opts.Prefix}.CONSUMER.RESET.{stream}.{consumer}",
+            request: seq == 0 ? null : new ConsumerResetRequest { Seq = seq },
+            cancellationToken);
+    }
+
     internal ValueTask<ConsumerInfo> CreateOrderedConsumerInternalAsync(
         string stream,
         NatsJSOrderedConsumerOpts opts,

src/NATS.Client.JetStream/NatsJSJsonSerializer.cs

@@ -32,6 +32,8 @@ public static class NatsJSJsonSerializer<T>
 [JsonSerializable(typeof(ConsumerNamesResponse))]
 [JsonSerializable(typeof(ConsumerPauseRequest))]
 [JsonSerializable(typeof(ConsumerPauseResponse))]
+[JsonSerializable(typeof(ConsumerResetRequest))]
+[JsonSerializable(typeof(ConsumerResetResponse))]
 [JsonSerializable(typeof(ConsumerUnpinRequest))]
 [JsonSerializable(typeof(ConsumerUnpinResponse))]
 [JsonSerializable(typeof(ErrorResponse))]

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<int>(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<int>(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<int>(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()
     {

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<ConsumerResetResponse> ResetConsumerAsync(string stream, string consumer, ulong seq = 0, CancellationToken cancellationToken = default) => throw new NotImplementedException();
+
         public ValueTask<AccountInfoResponse> GetAccountInfoAsync(CancellationToken cancellationToken = default) => throw new NotImplementedException();
 
         public ValueTask<PubAckResponse> PublishAsync<T>(string subject, T? data, INatsSerialize<T>? serializer = default, NatsJSPubOpts? opts = default, NatsHeaders? headers = default, CancellationToken cancellationToken = default) => throw new NotImplementedException();

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<ConsumerResetResponse> ResetConsumerAsync(string stream, string consumer, ulong seq = 0, CancellationToken cancellationToken = default) => throw new NotImplementedException();
+
         public ValueTask<AccountInfoResponse> GetAccountInfoAsync(CancellationToken cancellationToken = default) => throw new NotImplementedException();
 
         public ValueTask<PubAckResponse> PublishAsync<T>(string subject, T? data, INatsSerialize<T>? serializer = default, NatsJSPubOpts? opts = default, NatsHeaders? headers = default, CancellationToken cancellationToken = default) => throw new NotImplementedException();