| | | 1 | | using Microsoft.Extensions.Logging; |
| | | 2 | | using NATS.Client.Core; |
| | | 3 | | using NATS.Client.JetStream; |
| | | 4 | | using NATS.Client.JetStream.Models; |
| | | 5 | | using System.Collections.Concurrent; |
| | | 6 | | using System.Runtime.CompilerServices; |
| | | 7 | | |
| | | 8 | | namespace AsyncResponse.Transports.NATS; |
| | | 9 | | |
| | | 10 | | /// <summary>A worker/response message pulled from a JetStream consumer, decoupled from NATS client types for testabilit |
| | | 11 | | /// <param name="Subject">The subject the message was published to.</param> |
| | | 12 | | /// <param name="Payload">The raw JSON body.</param> |
| | | 13 | | /// <param name="Headers">The message headers (correlation id, etc.).</param> |
| | | 14 | | /// <param name="NumDelivered">How many times JetStream has delivered this message (1 on first delivery).</param> |
| | | 15 | | /// <param name="AckAsync">Acknowledges the message so it is not redelivered.</param> |
| | | 16 | | /// <param name="NakAsync">Negatively acknowledges the message, requesting redelivery after the given delay.</param> |
| | | 17 | | /// <param name="TermAsync">Terminates the message so JetStream stops redelivering it (used after dead-lettering).</para |
| | | 18 | | internal sealed record NatsJobDelivery( |
| | | 19 | | string Subject, |
| | | 20 | | string Payload, |
| | | 21 | | IReadOnlyDictionary<string, string> Headers, |
| | | 22 | | long NumDelivered, |
| | | 23 | | Func<ValueTask> AckAsync, |
| | | 24 | | Func<TimeSpan, ValueTask> NakAsync, |
| | | 25 | | Func<ValueTask> TermAsync) |
| | | 26 | | { |
| | | 27 | | /// <summary> |
| | | 28 | | /// Signals "working on it" (JetStream in-progress) so the server resets this delivery's |
| | | 29 | | /// AckWait window without settling it or bumping its delivery count. An init property with a |
| | | 30 | | /// no-op default rather than a positional parameter so out-of-package constructions stay |
| | | 31 | | /// source-compatible. The token is the batch's renewal cancellation: a heartbeat still in |
| | | 32 | | /// flight when the batch settles (or the subscriber stops) must abort with it, not hold the |
| | | 33 | | /// batch — the SDK call it wraps takes the token for exactly that. |
| | | 34 | | /// </summary> |
| | | 35 | | public Func<CancellationToken, ValueTask> ProgressAsync { get; init; } = static _ => ValueTask.CompletedTask; |
| | | 36 | | } |
| | | 37 | | |
| | | 38 | | /// <summary> |
| | | 39 | | /// Thin abstraction over the NATS JetStream operations the transport needs, confining the NATS.Net |
| | | 40 | | /// API surface to one place so the worker transport, dispatcher, and subscribers are unit-testable |
| | | 41 | | /// against a fake/mock. |
| | | 42 | | /// </summary> |
| | | 43 | | internal interface INatsJetStreamTransport |
| | | 44 | | { |
| | | 45 | | /// <summary> |
| | | 46 | | /// Creates the stream capturing <paramref name="subject"/> when it does not exist. An existing |
| | | 47 | | /// stream is verified, never rewritten. |
| | | 48 | | /// </summary> |
| | | 49 | | Task EnsureStreamAsync(string stream, string subject, long? maxMessages, CancellationToken cancellationToken); |
| | | 50 | | |
| | | 51 | | /// <summary> |
| | | 52 | | /// Ensures the dead-letter stream exists, with retention suited to a stream nothing consumes. |
| | | 53 | | /// </summary> |
| | | 54 | | Task EnsureDeadLetterStreamAsync(string stream, string subject, long? maxMessages, CancellationToken cancellationTok |
| | | 55 | | |
| | | 56 | | /// <summary>Idempotently creates or updates a durable explicit-ack consumer on <paramref name="stream"/>.</summary> |
| | | 57 | | Task EnsureConsumerAsync(string stream, string durable, TimeSpan ackWait, CancellationToken cancellationToken); |
| | | 58 | | |
| | | 59 | | /// <summary>Publishes <paramref name="payload"/> to <paramref name="subject"/> via JetStream and returns the assign |
| | | 60 | | Task<string> PublishAsync(string subject, string payload, IReadOnlyDictionary<string, string>? headers, Cancellation |
| | | 61 | | |
| | | 62 | | // The two fetch members carry throwing default implementations so out-of-package fakes that |
| | | 63 | | // never drive the subscriber read loop keep compiling; every implementation the subscribers |
| | | 64 | | // actually consume overrides them. |
| | | 65 | | |
| | | 66 | | /// <summary> |
| | | 67 | | /// Fetches up to <paramref name="maxMessages"/> already-available messages from the durable |
| | | 68 | | /// consumer and completes immediately (JetStream no-wait fetch) — the batch-drain half of the |
| | | 69 | | /// subscriber loop. |
| | | 70 | | /// </summary> |
| | | 71 | | IAsyncEnumerable<NatsJobDelivery> FetchNoWaitAsync(string stream, string durable, int maxMessages, CancellationToken |
| | 0 | 72 | | => throw new NotSupportedException($"{GetType()} does not implement {nameof(FetchNoWaitAsync)}."); |
| | | 73 | | |
| | | 74 | | /// <summary> |
| | | 75 | | /// Fetches up to <paramref name="maxMessages"/> messages, waiting up to |
| | | 76 | | /// <paramref name="expires"/> for them to arrive — the idle long-poll half of the subscriber |
| | | 77 | | /// loop. Completes without error when the wait expires with fewer messages. |
| | | 78 | | /// </summary> |
| | | 79 | | IAsyncEnumerable<NatsJobDelivery> FetchAsync(string stream, string durable, int maxMessages, TimeSpan expires, Cance |
| | 0 | 80 | | => throw new NotSupportedException($"{GetType()} does not implement {nameof(FetchAsync)}."); |
| | | 81 | | } |
| | | 82 | | |
| | | 83 | | /// <summary>Production <see cref="INatsJetStreamTransport"/> over a NATS <see cref="INatsJSContext"/>.</summary> |
| | | 84 | | internal sealed class NatsJetStreamTransportAdapter(INatsJSContext _jetStream, ILogger? _logger = null, int _streamRepli |
| | | 85 | | { |
| | | 86 | | // JetStream ApiError.ErrCode for "stream name already in use with a different configuration". |
| | | 87 | | private const int StreamNameInUseErrCode = 10058; |
| | | 88 | | |
| | | 89 | | /// <summary>Ensures the required resource exists.</summary> |
| | | 90 | | public Task EnsureStreamAsync(string stream, string subject, long? maxMessages, CancellationToken cancellationToken) |
| | | 91 | | { |
| | | 92 | | var config = new StreamConfig(stream, [subject]) |
| | | 93 | | { |
| | | 94 | | MaxMsgs = maxMessages ?? -1, |
| | | 95 | | // Work-queue retention removes each message once it is acked, so the stream only ever |
| | | 96 | | // holds the unprocessed backlog. Limits retention kept acked messages forever, letting |
| | | 97 | | // MaxMsgs eviction silently discard the oldest *unprocessed* jobs once the cap filled |
| | | 98 | | // up with already-acked traffic. |
| | | 99 | | Retention = StreamConfigRetention.Workqueue, |
| | | 100 | | // If the unprocessed backlog itself reaches MaxMsgs, refuse new publishes (a failed |
| | | 101 | | // PubAck the publisher's retry/exception path surfaces) instead of silently evicting |
| | | 102 | | // the oldest pending jobs. |
| | | 103 | | Discard = StreamConfigDiscard.New, |
| | | 104 | | NumReplicas = _streamReplicas |
| | | 105 | | }; |
| | | 106 | | return EnsureStreamAsync(stream, subject, config, retentionIsRequired: true, cancellationToken); |
| | | 107 | | } |
| | | 108 | | |
| | | 109 | | /// <summary>Ensures the dead-letter stream exists.</summary> |
| | | 110 | | public Task EnsureDeadLetterStreamAsync(string stream, string subject, long? maxMessages, CancellationToken cancella |
| | | 111 | | { |
| | | 112 | | // NOT the work-queue config above: nothing ever consumes (so nothing ever acks) the |
| | | 113 | | // dead-letter subject, which means work-queue retention removes nothing and Discard=New |
| | | 114 | | // then rejects every burial once MaxMsgs fills — each over-cap poison message NAK-looping |
| | | 115 | | // forever because its burial can never be accepted. Limits retention with Discard=Old |
| | | 116 | | // makes the DLQ a bounded evict-oldest archive, the same shape as Redis's MAXLEN-trimmed |
| | | 117 | | // dead-letter stream. |
| | | 118 | | var config = new StreamConfig(stream, [subject]) |
| | | 119 | | { |
| | | 120 | | MaxMsgs = maxMessages ?? -1, |
| | | 121 | | Retention = StreamConfigRetention.Limits, |
| | | 122 | | Discard = StreamConfigDiscard.Old, |
| | | 123 | | NumReplicas = _streamReplicas |
| | | 124 | | }; |
| | | 125 | | |
| | | 126 | | // Retention is NOT required here: a DLQ provisioned by an earlier build (work-queue |
| | | 127 | | // retention) cannot be changed in place — JetStream makes retention immutable — and it |
| | | 128 | | // still accepts burials until it fills; failing the whole subscriber over it would be |
| | | 129 | | // worse. Keep running and tell the operator how to migrate. |
| | | 130 | | return EnsureStreamAsync(stream, subject, config, retentionIsRequired: false, cancellationToken); |
| | | 131 | | } |
| | | 132 | | |
| | | 133 | | /// <summary> |
| | | 134 | | /// Creates the stream when it is missing and otherwise leaves it exactly as it is. This ran as |
| | | 135 | | /// a create-or-UPDATE with the minimal config above, and a JetStream update replaces the whole |
| | | 136 | | /// configuration: every subscriber start (and every first publish) reset whatever an operator |
| | | 137 | | /// had tuned on the live stream — replicas back to 1, max age / max bytes / max message size |
| | | 138 | | /// back to unlimited, the duplicate window back to its default. An existing stream is only |
| | | 139 | | /// checked for what this transport cannot work without; the rest of any drift is reported, |
| | | 140 | | /// never overwritten. |
| | | 141 | | /// </summary> |
| | | 142 | | private async Task EnsureStreamAsync(string stream, string subject, StreamConfig desired, bool retentionIsRequired, |
| | | 143 | | { |
| | | 144 | | var existing = await TryGetStreamConfigAsync(stream, cancellationToken).ConfigureAwait(false); |
| | | 145 | | if (existing is null) |
| | | 146 | | { |
| | | 147 | | try |
| | | 148 | | { |
| | | 149 | | // Creating with a configuration identical to the live one is a JetStream no-op, so |
| | | 150 | | // replicas of one deployment racing here all succeed. |
| | | 151 | | await _jetStream.CreateStreamAsync(desired, cancellationToken).ConfigureAwait(false); |
| | | 152 | | return; |
| | | 153 | | } |
| | | 154 | | catch (NatsJSApiException ex) when (ex.Error.ErrCode == StreamNameInUseErrCode) |
| | | 155 | | { |
| | | 156 | | // A peer configured differently (mid-rollout) won the creation race: from here on |
| | | 157 | | // it is an existing stream like any other. |
| | | 158 | | existing = await TryGetStreamConfigAsync(stream, cancellationToken).ConfigureAwait(false); |
| | | 159 | | if (existing is null) |
| | | 160 | | throw; |
| | | 161 | | } |
| | | 162 | | } |
| | | 163 | | |
| | | 164 | | VerifyExistingStream(stream, subject, existing, desired, retentionIsRequired); |
| | | 165 | | } |
| | | 166 | | |
| | | 167 | | private async Task<StreamConfig?> TryGetStreamConfigAsync(string stream, CancellationToken cancellationToken) |
| | | 168 | | { |
| | | 169 | | try |
| | | 170 | | { |
| | | 171 | | var info = await _jetStream.GetStreamAsync(stream, cancellationToken: cancellationToken).ConfigureAwait(fals |
| | | 172 | | return info.Info.Config; |
| | | 173 | | } |
| | | 174 | | catch (NatsJSApiException ex) when (ex.Error.Code == 404) |
| | | 175 | | { |
| | | 176 | | return null; // "stream not found" — the only answer that means the stream may be created |
| | | 177 | | } |
| | | 178 | | } |
| | | 179 | | |
| | | 180 | | private void VerifyExistingStream(string stream, string subject, StreamConfig existing, StreamConfig desired, bool r |
| | | 181 | | { |
| | | 182 | | // Nothing this transport publishes would be stored: every publish would fail with "no |
| | | 183 | | // response from stream" and the consumer would sit on an empty (or someone else's) stream. |
| | | 184 | | if (existing.Subjects is null || !existing.Subjects.Any(captured => SubjectCaptures(captured, subject))) |
| | | 185 | | { |
| | | 186 | | throw new InvalidOperationException( |
| | | 187 | | $"NATS stream '{stream}' already exists but does not capture subject '{subject}' " + |
| | | 188 | | $"(it captures: {(existing.Subjects is { Count: > 0 } subjects ? string.Join(", ", subjects) : "none")}) |
| | | 189 | | "An existing stream is never modified by this transport: add the subject to the stream, or configure a d |
| | | 190 | | } |
| | | 191 | | |
| | | 192 | | if (existing.Retention != desired.Retention) |
| | | 193 | | { |
| | | 194 | | if (retentionIsRequired) |
| | | 195 | | { |
| | | 196 | | // Without work-queue retention acked jobs are never removed: the stream fills with |
| | | 197 | | // finished work until MaxMsgs rejects every publish (or, with Discard=Old, evicts |
| | | 198 | | // jobs nobody has run yet). Retention cannot be changed on a live stream, so there |
| | | 199 | | // is nothing to repair here — fail, as the rejected update did before. |
| | | 200 | | throw new InvalidOperationException( |
| | | 201 | | $"NATS stream '{stream}' already exists with {existing.Retention} retention; this transport requires |
| | | 202 | | "and JetStream does not allow changing the retention policy of an existing stream. " + |
| | | 203 | | "Delete the stream (after draining it) so this host can recreate it, or configure a different stream |
| | | 204 | | } |
| | | 205 | | |
| | | 206 | | _logger?.LogWarning( |
| | | 207 | | "The NATS dead-letter stream {Stream} has {Retention} retention instead of limits retention (an existing |
| | | 208 | | "It keeps its current configuration; to migrate, delete and let this host recreate it (its messages are |
| | | 209 | | stream, |
| | | 210 | | existing.Retention); |
| | | 211 | | } |
| | | 212 | | |
| | | 213 | | if (existing.Discard != desired.Discard || existing.MaxMsgs != desired.MaxMsgs) |
| | | 214 | | { |
| | | 215 | | _logger?.LogWarning( |
| | | 216 | | "NATS stream {Stream} already exists with discard={Discard}, max_msgs={MaxMsgs}; this host is configured |
| | | 217 | | "An existing stream is never modified by this transport — apply the change to the stream yourself, or al |
| | | 218 | | stream, |
| | | 219 | | existing.Discard, |
| | | 220 | | existing.MaxMsgs, |
| | | 221 | | desired.Discard, |
| | | 222 | | desired.MaxMsgs); |
| | | 223 | | } |
| | | 224 | | } |
| | | 225 | | |
| | | 226 | | /// <summary>NATS subject matching: <c>*</c> stands for exactly one token, a trailing <c>></c> for one or more.</ |
| | | 227 | | internal static bool SubjectCaptures(string captured, string subject) |
| | | 228 | | { |
| | | 229 | | var capturedTokens = captured.Split('.'); |
| | | 230 | | var subjectTokens = subject.Split('.'); |
| | | 231 | | for (var i = 0; i < capturedTokens.Length; i++) |
| | | 232 | | { |
| | | 233 | | if (capturedTokens[i] == ">") |
| | | 234 | | return i < subjectTokens.Length; |
| | | 235 | | |
| | | 236 | | if (i >= subjectTokens.Length) |
| | | 237 | | return false; |
| | | 238 | | |
| | | 239 | | if (capturedTokens[i] != "*" && !string.Equals(capturedTokens[i], subjectTokens[i], StringComparison.Ordinal |
| | | 240 | | return false; |
| | | 241 | | } |
| | | 242 | | |
| | | 243 | | return capturedTokens.Length == subjectTokens.Length; |
| | | 244 | | } |
| | | 245 | | |
| | | 246 | | /// <summary>Ensures the required resource exists.</summary> |
| | | 247 | | public async Task EnsureConsumerAsync(string stream, string durable, TimeSpan ackWait, CancellationToken cancellatio |
| | | 248 | | { |
| | | 249 | | var config = new ConsumerConfig(durable) |
| | | 250 | | { |
| | | 251 | | DurableName = durable, |
| | | 252 | | AckPolicy = ConsumerConfigAckPolicy.Explicit, |
| | | 253 | | AckWait = ackWait, |
| | | 254 | | // Redelivery attempts are bounded by the dispatcher (via NumDelivered + Terminate), so the |
| | | 255 | | // consumer itself is left unlimited rather than silently swallowing the last attempt. |
| | | 256 | | MaxDeliver = -1 |
| | | 257 | | }; |
| | | 258 | | await _jetStream.CreateOrUpdateConsumerAsync(stream, config, cancellationToken).ConfigureAwait(false); |
| | | 259 | | } |
| | | 260 | | |
| | | 261 | | /// <summary>Publishes the supplied message.</summary> |
| | | 262 | | public async Task<string> PublishAsync(string subject, string payload, IReadOnlyDictionary<string, string>? headers, |
| | | 263 | | { |
| | | 264 | | var ack = await _jetStream.PublishAsync( |
| | | 265 | | subject, |
| | | 266 | | payload, |
| | | 267 | | headers: ToHeaders(headers), |
| | | 268 | | cancellationToken: cancellationToken).ConfigureAwait(false); |
| | | 269 | | |
| | | 270 | | EnsureAccepted(ack); |
| | | 271 | | |
| | | 272 | | return ack.Seq.ToString(); |
| | | 273 | | } |
| | | 274 | | |
| | | 275 | | /// <summary> |
| | | 276 | | /// Accepts a JetStream publish ack. Deliberately NOT <c>ack.EnsureSuccess()</c>: that treats |
| | | 277 | | /// <c>PubAck.Duplicate</c> as a failure and throws NatsJSDuplicateMessageException. |
| | | 278 | | /// Duplicate=true means JetStream already holds this Nats-Msg-Id — the SUCCESS case for the |
| | | 279 | | /// stable id the worker transport stamps outside its retry loop, precisely so a retry after a |
| | | 280 | | /// lost PubAck is deduplicated rather than enqueuing the same worker job twice. Throwing there |
| | | 281 | | /// burned the whole retry ladder (every attempt gets the same answer — it is a JetStream |
| | | 282 | | /// decision, not a blip) and reported a publish failure for a job that is queued and WILL run, |
| | | 283 | | /// so the caller re-published under a fresh id and the job executed twice. Only a real API |
| | | 284 | | /// error is a failure. |
| | | 285 | | /// </summary> |
| | | 286 | | internal static void EnsureAccepted(PubAckResponse ack) |
| | | 287 | | { |
| | | 288 | | if (ack.Error is not null) |
| | | 289 | | throw new NatsJSApiException(ack.Error); |
| | | 290 | | } |
| | | 291 | | |
| | | 292 | | /// <summary>Runs the FetchNoWaitAsync operation.</summary> |
| | | 293 | | public async IAsyncEnumerable<NatsJobDelivery> FetchNoWaitAsync( |
| | | 294 | | string stream, |
| | | 295 | | string durable, |
| | | 296 | | int maxMessages, |
| | | 297 | | [EnumeratorCancellation] CancellationToken cancellationToken) |
| | | 298 | | { |
| | | 299 | | var consumer = await GetConsumerAsync(stream, durable, cancellationToken).ConfigureAwait(false); |
| | | 300 | | var fetchOpts = new NatsJSFetchOpts { MaxMsgs = maxMessages }; |
| | | 301 | | |
| | | 302 | | await foreach (var message in consumer.FetchNoWaitAsync<string>(opts: fetchOpts, cancellationToken: cancellation |
| | | 303 | | yield return ToDelivery(message); |
| | | 304 | | } |
| | | 305 | | |
| | | 306 | | /// <summary>Runs the FetchAsync operation.</summary> |
| | | 307 | | public async IAsyncEnumerable<NatsJobDelivery> FetchAsync( |
| | | 308 | | string stream, |
| | | 309 | | string durable, |
| | | 310 | | int maxMessages, |
| | | 311 | | TimeSpan expires, |
| | | 312 | | [EnumeratorCancellation] CancellationToken cancellationToken) |
| | | 313 | | { |
| | | 314 | | var consumer = await GetConsumerAsync(stream, durable, cancellationToken).ConfigureAwait(false); |
| | | 315 | | var fetchOpts = new NatsJSFetchOpts { MaxMsgs = maxMessages, Expires = expires }; |
| | | 316 | | |
| | | 317 | | await foreach (var message in consumer.FetchAsync<string>(opts: fetchOpts, cancellationToken: cancellationToken) |
| | | 318 | | yield return ToDelivery(message); |
| | | 319 | | } |
| | | 320 | | |
| | | 321 | | // The consumer wrapper only carries names for building pull requests, so it stays valid across |
| | | 322 | | // subscriber rebuilds (EnsureConsumerAsync recreates the durable if it was deleted server-side) |
| | | 323 | | // and is cached to avoid a consumer-INFO round trip per fetch. Only a SUCCESSFUL lookup is |
| | | 324 | | // cached, so a transient failure is not replayed forever. |
| | | 325 | | private readonly ConcurrentDictionary<(string Stream, string Durable), INatsJSConsumer> _consumers = new(); |
| | | 326 | | |
| | | 327 | | private async ValueTask<INatsJSConsumer> GetConsumerAsync(string stream, string durable, CancellationToken cancellat |
| | | 328 | | { |
| | | 329 | | if (_consumers.TryGetValue((stream, durable), out var cached)) |
| | | 330 | | return cached; |
| | | 331 | | |
| | | 332 | | var consumer = await _jetStream.GetConsumerAsync(stream, durable, cancellationToken).ConfigureAwait(false); |
| | | 333 | | return _consumers.GetOrAdd((stream, durable), consumer); |
| | | 334 | | } |
| | | 335 | | |
| | | 336 | | private static NatsJobDelivery ToDelivery(INatsJSMsg<string> message) |
| | | 337 | | { |
| | | 338 | | var numDelivered = (long)(message.Metadata?.NumDelivered ?? 1); |
| | | 339 | | var captured = message; |
| | | 340 | | |
| | | 341 | | return new NatsJobDelivery( |
| | | 342 | | captured.Subject, |
| | | 343 | | captured.Data ?? string.Empty, |
| | | 344 | | FromHeaders(captured.Headers), |
| | | 345 | | numDelivered, |
| | | 346 | | () => captured.AckAsync(cancellationToken: CancellationToken.None), |
| | | 347 | | delay => captured.NakAsync(delay: delay, cancellationToken: CancellationToken.None), |
| | | 348 | | () => captured.AckTerminateAsync(cancellationToken: CancellationToken.None)) |
| | | 349 | | { |
| | | 350 | | // Unlike the settlements above (deliberately uncancelable: a settlement decision |
| | | 351 | | // already taken must reach the server), a progress heartbeat is advisory — the |
| | | 352 | | // renewal loop's token cancels one that stalls, so the batch never waits on it. |
| | | 353 | | ProgressAsync = cancellationToken => captured.AckProgressAsync(cancellationToken: cancellationToken) |
| | | 354 | | }; |
| | | 355 | | } |
| | | 356 | | |
| | | 357 | | private static NatsHeaders? ToHeaders(IReadOnlyDictionary<string, string>? headers) |
| | | 358 | | { |
| | | 359 | | if (headers is null || headers.Count == 0) |
| | | 360 | | return null; |
| | | 361 | | |
| | | 362 | | var natsHeaders = new NatsHeaders(); |
| | | 363 | | foreach (var (key, value) in headers) |
| | | 364 | | natsHeaders[key] = value; |
| | | 365 | | return natsHeaders; |
| | | 366 | | } |
| | | 367 | | |
| | | 368 | | private static IReadOnlyDictionary<string, string> FromHeaders(NatsHeaders? headers) |
| | | 369 | | { |
| | | 370 | | if (headers is null || headers.Count == 0) |
| | | 371 | | return EmptyHeaders; |
| | | 372 | | |
| | | 373 | | var result = new Dictionary<string, string>(headers.Count, StringComparer.OrdinalIgnoreCase); |
| | | 374 | | foreach (var key in headers.Keys) |
| | | 375 | | result[key] = headers[key].ToString(); |
| | | 376 | | return result; |
| | | 377 | | } |
| | | 378 | | |
| | | 379 | | private static readonly IReadOnlyDictionary<string, string> EmptyHeaders = |
| | | 380 | | new Dictionary<string, string>(0, StringComparer.OrdinalIgnoreCase); |
| | | 381 | | } |
| | | 382 | | |
| | | 383 | | /// <summary>Bounded exponential-backoff retry for transient NATS failures, mirroring the other transports.</summary> |
| | | 384 | | internal static class NatsTransportRetry |
| | | 385 | | { |
| | | 386 | | /// <summary>Runs this background operation until cancellation is requested.</summary> |
| | | 387 | | public static Task<T> ExecuteAsync<T>( |
| | | 388 | | Func<CancellationToken, Task<T>> action, |
| | | 389 | | int maxAttempts, |
| | | 390 | | TimeSpan baseDelay, |
| | | 391 | | TimeSpan maxDelay, |
| | | 392 | | CancellationToken cancellationToken) |
| | | 393 | | => AsyncResponseRetry.ExecuteAsync(action, IsTransient, maxAttempts, baseDelay, maxDelay, cancellationToken); |
| | | 394 | | |
| | | 395 | | /// <summary>Runs the IsTransient operation.</summary> |
| | | 396 | | public static bool IsTransient(Exception exception) |
| | | 397 | | { |
| | | 398 | | // A JetStream API request that the server ANSWERED with an error is a decision, not a |
| | | 399 | | // blip: "stream name already in use", "consumer config would change an immutable field", |
| | | 400 | | // "no permission". Those repeat identically on every attempt, so retrying only delays the |
| | | 401 | | // report — unless the server itself said it was temporarily unable (5xx, e.g. 503 while a |
| | | 402 | | // meta-leader election settles). Everything else in the NatsException family — no |
| | | 403 | | // responders, no API response, connection loss — is the transient case. |
| | | 404 | | if (exception is NatsJSApiException api) |
| | | 405 | | return api.Error.Code >= 500; |
| | | 406 | | |
| | | 407 | | return exception is NatsException or TimeoutException && exception is not OperationCanceledException; |
| | | 408 | | } |
| | | 409 | | } |