Skip to content

Add TermWithReason support to AckTerminateAsync #1048

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mtmk merged 2 commits into from
Feb 26, 2026
Merged

Add TermWithReason support to AckTerminateAsync #1048

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
be4d997
Add TermWithReason support to AckTerminateAsync
mtmk Jan 22, 2026
48d1fe3
use overload to not break compat
mtmk Jan 23, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
179 changes: 86 additions & 93 deletions src/NATS.Client.JetStream/NatsJSMsg.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@ namespace NATS.Client.JetStream;

/// <summary>
/// This interface provides an optional contract when passing
/// messages to processing methods which is usually helpful in
/// messages to processing methods, which is usually helpful in
/// creating test doubles in unit testing.
/// </summary>
/// <remarks>
Expand All @@ -30,12 +30,12 @@ namespace NATS.Client.JetStream;
public interface INatsJSMsg<out T> : INatsMsg
{
/// <summary>
/// Subject of the user message.
/// Get subject of the user message.
/// </summary>
string Subject { get; }

/// <summary>
/// Message size in bytes.
/// Get message size in bytes.
/// </summary>
/// <remarks>
/// Message size is calculated using the same method NATS server uses:
Expand All @@ -46,12 +46,12 @@ public interface INatsJSMsg<out T> : INatsMsg
int Size { get; }

/// <summary>
/// Deserialized user data.
/// Get deserialized user data.
/// </summary>
T? Data { get; }

/// <summary>
/// The connection messages was delivered on.
/// Get the connection messages were delivered on.
/// </summary>
INatsConnection? Connection { get; }

Expand Down Expand Up @@ -80,15 +80,15 @@ public interface INatsJSMsg<out T> : INatsMsg
/// <param name="cancellationToken">A <see cref="CancellationToken"/> used to cancel the command.</param>
/// <returns>A <see cref="ValueTask"/> that represents the asynchronous send operation.</returns>
[Obsolete("ReplyAsync is not valid when using JetStream. The reply message will never reach the Requestor as NATS will reply to all JetStream publish by default.")]
ValueTask ReplyAsync(NatsHeaders? headers = default, string? replyTo = default, NatsPubOpts? opts = default, CancellationToken cancellationToken = default);
ValueTask ReplyAsync(NatsHeaders? headers = null, string? replyTo = null, NatsPubOpts? opts = null, CancellationToken cancellationToken = default);

/// <summary>
/// Acknowledges the message was completely handled.
/// </summary>
/// <param name="opts">Ack options.</param>
/// <param name="cancellationToken">A <see cref="CancellationToken"/> used to cancel the call.</param>
/// <returns>A <see cref="ValueTask"/> representing the async call.</returns>
ValueTask AckAsync(AckOpts? opts = default, CancellationToken cancellationToken = default);
ValueTask AckAsync(AckOpts? opts = null, CancellationToken cancellationToken = default);

/// <summary>
/// Signals that the message will not be processed now and processing can move onto the next message.
Expand All @@ -101,7 +101,7 @@ public interface INatsJSMsg<out T> : INatsMsg
/// Messages rejected using <c>-NAK</c> will be resent by the NATS JetStream server after the configured timeout
/// or the delay parameter if it's specified.
/// </remarks>
ValueTask NakAsync(AckOpts? opts = default, TimeSpan delay = default, CancellationToken cancellationToken = default);
ValueTask NakAsync(AckOpts? opts = null, TimeSpan delay = default, CancellationToken cancellationToken = default);

/// <summary>
/// Indicates that work is ongoing and the wait period should be extended.
Expand All @@ -112,22 +112,31 @@ public interface INatsJSMsg<out T> : INatsMsg
/// <remarks>
/// <para>
/// Time period is defined by the consumer's <c>ack_wait</c> configuration on the server which is
/// defined as how long to allow messages to remain un-acknowledged before attempting redelivery.
/// defined as how long to allow messages to remain unacknowledged before attempting redelivery.
/// </para>
/// <para>
/// This message must be sent before the <c>ack_wait</c> period elapses. The period should be extended
/// by another amount of time equal to <c>ack_wait</c> by the NATS JetStream server.
/// </para>
/// </remarks>
ValueTask AckProgressAsync(AckOpts? opts = default, CancellationToken cancellationToken = default);
ValueTask AckProgressAsync(AckOpts? opts = null, CancellationToken cancellationToken = default);

/// <summary>
/// Instructs the server to stop redelivery of the message without acknowledging it as successfully processed.
/// </summary>
/// <param name="opts">Ack options.</param>
/// <param name="cancellationToken">A <see cref="CancellationToken"/> used to cancel the call.</param>
/// <returns>A <see cref="ValueTask"/> representing the async call.</returns>
ValueTask AckTerminateAsync(AckOpts? opts = default, CancellationToken cancellationToken = default);
ValueTask AckTerminateAsync(AckOpts? opts = null, CancellationToken cancellationToken = default);

/// <summary>
/// Instructs the server to stop redelivery of the message without acknowledging it as successfully processed.
/// </summary>
/// <param name="reason">Optional reason for termination, included in JetStream advisory events. Requires NATS Server 2.10.4+.</param>
/// <param name="opts">Ack options.</param>
/// <param name="cancellationToken">A <see cref="CancellationToken"/> used to cancel the call.</param>
/// <returns>A <see cref="ValueTask"/> representing the async call.</returns>
ValueTask AckTerminateAsync(string reason, AckOpts? opts = null, CancellationToken cancellationToken = default);
}

/// <summary>
Expand All @@ -147,45 +156,25 @@ public NatsJSMsg(NatsMsg<T> msg, INatsJSContext context)
_replyToDateTimeAndSeq = new Lazy<NatsJSMsgMetadata?>(() => ReplyToDateTimeAndSeq.Parse(msg.ReplyTo));
}

/// <summary>
/// Subject of the user message.
/// </summary>
/// <inheritdoc />
public string Subject => _msg.Subject;

/// <summary>
/// Message size in bytes.
/// </summary>
/// <remarks>
/// Message size is calculated using the same method NATS server uses:
/// <code lang="C#">
/// int size = subject.Length + replyTo.Length + headers.Length + payload.Length;
/// </code>
/// </remarks>
/// <inheritdoc />
public int Size => _msg.Size;

/// <summary>
/// Headers of the user message if set.
/// </summary>
/// <inheritdoc />
public NatsHeaders? Headers => _msg.Headers;

/// <summary>
/// Deserialized user data.
/// </summary>
/// <inheritdoc />
public T? Data => _msg.Data;

/// <summary>
/// The connection messages was delivered on.
/// </summary>
/// <inheritdoc />
public INatsConnection? Connection => _msg.Connection;

/// <summary>
/// Additional metadata about the message.
/// </summary>
/// <inheritdoc />
public NatsJSMsgMetadata? Metadata => _replyToDateTimeAndSeq.Value;

/// <summary>
/// The reply subject that subscribers can use to send a response back to the publisher/requester.
/// </summary>
/// <inheritdoc />
public string? ReplyTo => _msg.ReplyTo;

/// <inheritdoc />
Expand All @@ -196,40 +185,18 @@ public NatsJSMsg(NatsMsg<T> msg, INatsJSContext context)
/// <inheritdoc />
public void EnsureSuccess() => _msg.EnsureSuccess();

/// <summary>
/// Reply with an empty message.
/// </summary>
/// <param name="headers">Optional message headers.</param>
/// <param name="replyTo">Optional reply-to subject.</param>
/// <param name="opts">A <see cref="NatsPubOpts"/> for publishing options.</param>
/// <param name="cancellationToken">A <see cref="CancellationToken"/> used to cancel the command.</param>
/// <returns>A <see cref="ValueTask"/> that represents the asynchronous send operation.</returns>
/// <inheritdoc />
[Obsolete("ReplyAsync is not valid when using JetStream. The reply message will never reach the Requestor as NATS will reply to all JetStream publish by default.")]
public ValueTask ReplyAsync(NatsHeaders? headers = default, string? replyTo = default, NatsPubOpts? opts = default, CancellationToken cancellationToken = default) =>
public ValueTask ReplyAsync(NatsHeaders? headers = null, string? replyTo = null, NatsPubOpts? opts = null, CancellationToken cancellationToken = default) =>
_msg.ReplyAsync(headers, replyTo, opts, cancellationToken);

/// <summary>
/// Acknowledges the message was completely handled.
/// </summary>
/// <param name="opts">Ack options.</param>
/// <param name="cancellationToken">A <see cref="CancellationToken"/> used to cancel the call.</param>
/// <returns>A <see cref="ValueTask"/> representing the async call.</returns>
public ValueTask AckAsync(AckOpts? opts = default, CancellationToken cancellationToken = default) => SendAckAsync(NatsJSConstants.Ack, opts, cancellationToken);
/// <inheritdoc />
public ValueTask AckAsync(AckOpts? opts = null, CancellationToken cancellationToken = default) => SendAckAsync(NatsJSConstants.Ack, opts, cancellationToken);

/// <summary>
/// Signals that the message will not be processed now and processing can move onto the next message.
/// </summary>
/// <param name="delay">Delay redelivery of the message.</param>
/// <param name="opts">Ack options.</param>
/// <param name="cancellationToken">A <see cref="CancellationToken"/> used to cancel the call.</param>
/// <returns>A <see cref="ValueTask"/> representing the async call.</returns>
/// <remarks>
/// Messages rejected using <c>-NAK</c> will be resent by the NATS JetStream server after the configured timeout
/// or the delay parameter if it's specified.
/// </remarks>
public ValueTask NakAsync(AckOpts? opts = default, TimeSpan delay = default, CancellationToken cancellationToken = default)
/// <inheritdoc />
public ValueTask NakAsync(AckOpts? opts = null, TimeSpan delay = default, CancellationToken cancellationToken = default)
{
if (delay == default)
if (delay == TimeSpan.Zero)
{
return SendAckAsync(NatsJSConstants.Nak, opts, cancellationToken);
}
Expand All @@ -240,33 +207,52 @@ public ValueTask NakAsync(AckOpts? opts = default, TimeSpan delay = default, Can
}
}

/// <summary>
/// Indicates that work is ongoing and the wait period should be extended.
/// </summary>
/// <param name="opts">Ack options.</param>
/// <param name="cancellationToken">A <see cref="CancellationToken"/> used to cancel the call.</param>
/// <returns>A <see cref="ValueTask"/> representing the async call.</returns>
/// <remarks>
/// <para>
/// Time period is defined by the consumer's <c>ack_wait</c> configuration on the server which is
/// defined as how long to allow messages to remain un-acknowledged before attempting redelivery.
/// </para>
/// <para>
/// This message must be sent before the <c>ack_wait</c> period elapses. The period should be extended
/// by another amount of time equal to <c>ack_wait</c> by the NATS JetStream server.
/// </para>
/// </remarks>
public ValueTask AckProgressAsync(AckOpts? opts = default, CancellationToken cancellationToken = default) => SendAckAsync(NatsJSConstants.AckProgress, opts, cancellationToken);
/// <inheritdoc />
public ValueTask AckProgressAsync(AckOpts? opts = null, CancellationToken cancellationToken = default) => SendAckAsync(NatsJSConstants.AckProgress, opts, cancellationToken);

/// <summary>
/// Instructs the server to stop redelivery of the message without acknowledging it as successfully processed.
/// </summary>
/// <param name="opts">Ack options.</param>
/// <param name="cancellationToken">A <see cref="CancellationToken"/> used to cancel the call.</param>
/// <returns>A <see cref="ValueTask"/> representing the async call.</returns>
public ValueTask AckTerminateAsync(AckOpts? opts = default, CancellationToken cancellationToken = default) => SendAckAsync(NatsJSConstants.AckTerminate, opts, cancellationToken);
/// <inheritdoc />
public ValueTask AckTerminateAsync(AckOpts? opts = null, CancellationToken cancellationToken = default)
=> AckTerminateInternalAsync(opts, null, cancellationToken);

/// <inheritdoc />
public ValueTask AckTerminateAsync(string reason, AckOpts? opts = null, CancellationToken cancellationToken = default)
=> AckTerminateInternalAsync(opts, reason, cancellationToken);

private ValueTask AckTerminateInternalAsync(AckOpts? opts, string? reason, CancellationToken cancellationToken)
{
if (string.IsNullOrEmpty(reason))
{
return SendAckAsync(NatsJSConstants.AckTerminate, opts, cancellationToken);
}

return AckTerminateWithReasonAsync(reason!, opts, cancellationToken);
}

private async ValueTask SendAckAsync(ReadOnlySequence<byte> payload, AckOpts? opts = default, CancellationToken cancellationToken = default)
private async ValueTask AckTerminateWithReasonAsync(string reason, AckOpts? opts, CancellationToken cancellationToken)
{
var reasonByteCount = Encoding.ASCII.GetByteCount(reason);
var totalLength = 6 + reasonByteCount; // `+TERM ` is 6 bytes

var buffer = ArrayPool<byte>.Shared.Rent(totalLength);
try
{
#if NETSTANDARD2_0
Buffer.BlockCopy(NatsJSMsgConstants.TermPrefix, 0, buffer, 0, 6);
Encoding.ASCII.GetBytes(reason, 0, reason.Length, buffer, 6);
#else
"+TERM "u8.CopyTo(buffer.AsSpan());
Encoding.ASCII.GetBytes(reason.AsSpan(), buffer.AsSpan(6));
#endif

await SendAckAsync(new ReadOnlySequence<byte>(buffer, 0, totalLength), opts, cancellationToken);
}
finally
{
ArrayPool<byte>.Shared.Return(buffer);
}
}

private async ValueTask SendAckAsync(ReadOnlySequence<byte> payload, AckOpts? opts = null, CancellationToken cancellationToken = default)
{
CheckPreconditions();

Expand Down Expand Up @@ -298,7 +284,7 @@ await _msg.ReplyAsync(
[MemberNotNull(nameof(ReplyTo))]
private void CheckPreconditions()
{
if (Connection == default)
if (Connection == null)
{
throw new NatsException("unable to send acknowledgment; message did not originate from a consumer");
}
Expand All @@ -323,3 +309,10 @@ public readonly record struct AckOpts
/// </summary>
public bool? DoubleAck { get; init; }
}

#if NETSTANDARD2_0
internal static class NatsJSMsgConstants
{
internal static readonly byte[] TermPrefix = "+TERM "u8.ToArray();
}
#endif
74 changes: 74 additions & 0 deletions tests/NATS.Client.JetStream.Tests/DoubleAckNakDelayTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,4 +91,78 @@ public async Task Delay_nak_received_messages(NatsRequestReplyMode mode)
Assert.Fail("No message received");
}
}

[Theory]
[InlineData(NatsRequestReplyMode.Direct)]
[InlineData(NatsRequestReplyMode.SharedInbox)]
public async Task Terminate_without_reason(NatsRequestReplyMode mode)
{
var proxy = _server.CreateProxy();
await using var nats = proxy.CreateNatsConnection(mode);
await nats.ConnectRetryAsync();
var prefix = _server.GetNextId();

var js = new NatsJSContext(nats);

var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30));

await js.CreateStreamAsync($"{prefix}s1", [$"{prefix}s1.*"], cts.Token);
var consumer = await js.CreateOrUpdateConsumerAsync($"{prefix}s1", $"{prefix}c1", cancellationToken: cts.Token);

var ack = await js.PublishAsync($"{prefix}s1.foo", 42, cancellationToken: cts.Token);
ack.EnsureSuccess();
var next = await consumer.NextAsync<int>(cancellationToken: cts.Token);
if (next is { } msg)
{
await msg.AckTerminateAsync(cancellationToken: cts.Token);
Assert.Equal(42, msg.Data);

await Retry.Until("seen TERM", () => proxy.Frames.Any(f => f.Message.StartsWith("PUB $JS.ACK")));

var termFrame = proxy.Frames.Single(f => f.Message.StartsWith("PUB $JS.ACK"));

Assert.Matches(@"\+TERM\s*$", termFrame.Message);
}
else
{
Assert.Fail("No message received");
}
}

[Theory]
[InlineData(NatsRequestReplyMode.Direct)]
[InlineData(NatsRequestReplyMode.SharedInbox)]
public async Task Terminate_with_reason(NatsRequestReplyMode mode)
{
var proxy = _server.CreateProxy();
await using var nats = proxy.CreateNatsConnection(mode);
await nats.ConnectRetryAsync();
var prefix = _server.GetNextId();

var js = new NatsJSContext(nats);

var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30));

await js.CreateStreamAsync($"{prefix}s1", [$"{prefix}s1.*"], cts.Token);
var consumer = await js.CreateOrUpdateConsumerAsync($"{prefix}s1", $"{prefix}c1", cancellationToken: cts.Token);

var ack = await js.PublishAsync($"{prefix}s1.foo", 42, cancellationToken: cts.Token);
ack.EnsureSuccess();
var next = await consumer.NextAsync<int>(cancellationToken: cts.Token);
if (next is { } msg)
{
await msg.AckTerminateAsync(reason: "test failure reason", cancellationToken: cts.Token);
Assert.Equal(42, msg.Data);

await Retry.Until("seen TERM", () => proxy.Frames.Any(f => f.Message.StartsWith("PUB $JS.ACK")));

var termFrame = proxy.Frames.Single(f => f.Message.StartsWith("PUB $JS.ACK"));

Assert.Contains("+TERM test failure reason", termFrame.Message);
}
else
{
Assert.Fail("No message received");
}
}
}
Loading