Skip to content

Fix Object Store and JetStream publish retry logic #937

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 5 commits into from
Aug 28, 2025
Merged

Fix Object Store and JetStream publish retry logic #937

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
9e3d579
Fix JetStream publis retry logic
mtmk Aug 26, 2025
7fab153
Fix JetStream publish retry behavior and add tests
mtmk Aug 27, 2025
561aaa5
Update tests/NATS.Client.JetStream.Tests/PublishTest.cs
mtmk Aug 27, 2025
1960ad2
Fix JetStream publish retry condition
mtmk Aug 27, 2025
32f72fb
Fix object store publish to disable retries
mtmk Aug 28, 2025
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
19 changes: 12 additions & 7 deletions src/NATS.Client.JetStream/NatsJSContext.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -170,7 +170,6 @@ public async ValueTask<NatsResult<PubAckResponse>> TryPublishAsync<T>(
{
if (Connection.Opts.RequestReplyMode == NatsRequestReplyMode.Direct)
{
var noReply = false;
NatsMsg<PubAckResponse> msg;
try
{
Expand All @@ -186,15 +185,14 @@ public async ValueTask<NatsResult<PubAckResponse>> TryPublishAsync<T>(
}
catch (NatsNoReplyException)
{
noReply = true;
msg = default;
return NatsJSPublishNoResponseException.Default;
}
catch (Exception ex)
catch (NatsException ex)
{
return ex;
}

if (noReply || msg.HasNoResponders)
if (msg.HasNoResponders)
{
_logger.LogDebug(NatsJSLogEvents.PublishNoResponseRetry, "No response received, retrying {RetryCount}/{RetryMax}", i + 1, retryMax);
await Task.Delay(retryWait, cancellationToken);
Expand Down Expand Up @@ -228,13 +226,15 @@ public async ValueTask<NatsResult<PubAckResponse>> TryPublishAsync<T>(
cancellationToken)
.ConfigureAwait(false);

var hasNoResponders = false;
await foreach (var msg in sub.Msgs.ReadAllAsync(cancellationToken).ConfigureAwait(false))
{
// If JetStream is disabled, a no responders error will be returned.
// No responders error might also happen when reconnecting to cluster.
// We should retry in those cases.
if (msg.HasNoResponders)
{
hasNoResponders = true;
break;
}
else if (msg.Data == null)
Expand All @@ -245,16 +245,21 @@ public async ValueTask<NatsResult<PubAckResponse>> TryPublishAsync<T>(
return msg.Data;
}

if (i < retryMax)
// Only retry if there were 503 no responders error
if (hasNoResponders)
{
_logger.LogDebug(NatsJSLogEvents.PublishNoResponseRetry, "No response received, retrying {RetryCount}/{RetryMax}", i + 1, retryMax);
await Task.Delay(retryWait, cancellationToken);
}
else
{
break;
}
}

// We throw a specific exception here for convenience so that the caller doesn't
// have to check for the exception message etc.
return new NatsJSPublishNoResponseException();
return NatsJSPublishNoResponseException.Default;
}

public async ValueTask<NatsJSPublishConcurrentFuture> PublishConcurrentAsync<T>(
Expand Down
2 changes: 2 additions & 0 deletions src/NATS.Client.JetStream/NatsJSException.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -94,6 +94,8 @@ public NatsJSApiException(ApiError error)

public class NatsJSPublishNoResponseException : NatsJSException
{
public static readonly NatsJSPublishNoResponseException Default = new();

public NatsJSPublishNoResponseException()
: base("No response received from the server")
{
Expand Down
6 changes: 4 additions & 2 deletions src/NATS.Client.ObjectStore/NatsObjStore.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,8 @@ public class NatsObjStore : INatsObjStore
private const string NatsRollup = "Nats-Rollup";
private const string RollupSubject = "sub";

// Disable retries as we want to fail fast and avoid duplicate chunks
private readonly NatsJSPubOpts _natsJSPubOpts = new() { RetryAttempts = 1 };
private readonly NatsObjContext _objContext;
private readonly INatsJSStream _stream;

Expand Down Expand Up @@ -255,7 +257,7 @@ public async ValueTask<ObjectMetadata> PutAsync(ObjectMetadata meta, Stream stre
var buffer = memoryOwner.Slice(0, currentChunkSize);

// Chunks
var ack = await JetStreamContext.PublishAsync(GetChunkSubject(nuid), buffer, serializer: NatsRawSerializer<NatsMemoryOwner<byte>>.Default, cancellationToken: cancellationToken);
var ack = await JetStreamContext.PublishAsync(GetChunkSubject(nuid), buffer, serializer: NatsRawSerializer<NatsMemoryOwner<byte>>.Default, opts: _natsJSPubOpts, cancellationToken: cancellationToken);
ack.EnsureSuccess();

if (eof)
Expand Down Expand Up @@ -608,7 +610,7 @@ public async ValueTask DeleteAsync(string key, CancellationToken cancellationTok
private async ValueTask PublishMeta(ObjectMetadata meta, CancellationToken cancellationToken)
{
var natsRollupHeaders = new NatsHeaders { { NatsRollup, RollupSubject } };
var ack = await JetStreamContext.PublishAsync(GetMetaSubject(meta.Name), meta, serializer: NatsObjJsonSerializer<ObjectMetadata>.Default, headers: natsRollupHeaders, cancellationToken: cancellationToken);
var ack = await JetStreamContext.PublishAsync(GetMetaSubject(meta.Name), meta, serializer: NatsObjJsonSerializer<ObjectMetadata>.Default, headers: natsRollupHeaders, opts: _natsJSPubOpts, cancellationToken: cancellationToken);
ack.EnsureSuccess();
}

Expand Down
79 changes: 54 additions & 25 deletions tests/NATS.Client.JetStream.Tests/PublishTest.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -183,7 +183,7 @@ public async Task Publish_test(NatsRequestReplyMode mode)
[Theory]
[InlineData(NatsRequestReplyMode.Direct)]
[InlineData(NatsRequestReplyMode.SharedInbox)]
public async Task Publish_retry_test(NatsRequestReplyMode mode)
public async Task Publish_retry_fails_when_no_response_is_received_from_server(NatsRequestReplyMode mode)
{
var retryCount = 0;
var logger = new InMemoryTestLoggerFactory(LogLevel.Debug, log =>
Expand Down Expand Up @@ -245,44 +245,73 @@ public async Task Publish_retry_test(NatsRequestReplyMode mode)
await Retry.Until("ack received", () => proxy.Frames.Any(f => ackRegex.IsMatch(f.Message)));
}

// Publish fails once but succeeds after retry
// Publish fails without retry. We only retry on 503 and not receiving any response
// must not trigger a retry since we don't know for sure the publish is failed.
{
await proxy.FlushFramesAsync(nats, clear: true, cts.Token);
Interlocked.Exchange(ref retryCount, 0);
Interlocked.Exchange(ref swallowAcksCount, 1);

var ack = await js.PublishAsync($"{prefix}s1.foo", 1, opts: new NatsJSPubOpts { RetryAttempts = 2 }, cancellationToken: cts.Token);
ack.EnsureSuccess();
await Assert.ThrowsAnyAsync<NatsJSPublishNoResponseException>(async () =>
{
var ack = await js.PublishAsync($"{prefix}s1.foo", 1, opts: new NatsJSPubOpts { RetryAttempts = 2 }, cancellationToken: cts.Token);
ack.EnsureSuccess();
});

Assert.Equal(1, Volatile.Read(ref retryCount));
await Retry.Until("ack received", () => proxy.Frames.Count(f => ackRegex.IsMatch(f.Message)) == 2, timeout: TimeSpan.FromSeconds(20));
Assert.Equal(0, Volatile.Read(ref retryCount));
}
}

// Publish fails twice but succeeds after a third retry when attempts is 3
[Theory]
[InlineData(NatsRequestReplyMode.Direct)]
[InlineData(NatsRequestReplyMode.SharedInbox)]
public async Task Publish_retry_on_503(NatsRequestReplyMode mode)
{
await using var server = await NatsServerProcess.StartAsync(withJs: false);
var retryCount = 0;
var logger = new InMemoryTestLoggerFactory(LogLevel.Debug, log =>
{
await proxy.FlushFramesAsync(nats, clear: true, cts.Token);
Interlocked.Exchange(ref retryCount, 0);
Interlocked.Exchange(ref swallowAcksCount, 2);
if (log is { LogLevel: LogLevel.Debug } && log.EventId == NatsJSLogEvents.PublishNoResponseRetry)
{
Interlocked.Increment(ref retryCount);
}
});

var ack = await js.PublishAsync($"{prefix}s1.foo", 1, opts: new NatsJSPubOpts { RetryAttempts = 3 }, cancellationToken: cts.Token);
ack.EnsureSuccess();
await using var nats = new NatsConnection(new NatsOpts
{
Url = server.Url,
ConnectTimeout = TimeSpan.FromSeconds(10),
RequestTimeout = TimeSpan.FromSeconds(3), // give enough time for retries to avoid NatsJSPublishNoResponseExceptions
LoggerFactory = logger,
RequestReplyMode = mode,
});

Assert.Equal(2, Volatile.Read(ref retryCount));
await Retry.Until("ack received", () => proxy.Frames.Count(f => ackRegex.IsMatch(f.Message)) == 3, timeout: TimeSpan.FromSeconds(20));
}
var cts = new CancellationTokenSource(TimeSpan.FromSeconds(60));
var js = new NatsJSContext(nats);

// Publish fails even after two retries
{
await proxy.FlushFramesAsync(nats, clear: true, cts.Token);
Interlocked.Exchange(ref retryCount, 0);
Interlocked.Exchange(ref swallowAcksCount, 2);
// Default is two attempts
await Assert.ThrowsAnyAsync<NatsJSPublishNoResponseException>(async () => await js.PublishAsync($"foo", 1, cancellationToken: cts.Token));
Assert.Equal(2, Volatile.Read(ref retryCount));

await Assert.ThrowsAsync<NatsJSPublishNoResponseException>(async () =>
await js.PublishAsync($"{prefix}s1.foo", 1, opts: new NatsJSPubOpts { RetryAttempts = 2 }, cancellationToken: cts.Token));
// Set to multiple attempts
var attempts = 5;
Interlocked.Exchange(ref retryCount, 0);
await Assert.ThrowsAnyAsync<NatsJSPublishNoResponseException>(async () =>
{
var opts = new NatsJSPubOpts { RetryAttempts = attempts };
await js.PublishAsync($"foo", 1, opts: opts, cancellationToken: cts.Token);
});
Assert.Equal(attempts, Volatile.Read(ref retryCount));

Assert.Equal(2, Volatile.Read(ref retryCount));
await Retry.Until("ack received", () => proxy.Frames.Count(f => ackRegex.IsMatch(f.Message)) == 2, timeout: TimeSpan.FromSeconds(20));
}
// Disable retries
attempts = 1;
Interlocked.Exchange(ref retryCount, 0);
await Assert.ThrowsAnyAsync<NatsJSPublishNoResponseException>(async () =>
{
var opts = new NatsJSPubOpts { RetryAttempts = attempts };
await js.PublishAsync($"foo", 1, opts: opts, cancellationToken: cts.Token);
});
Assert.Equal(attempts, Volatile.Read(ref retryCount));
}

[Theory]
Expand Down
Loading