< Summary - AsyncResponse (Release / net8.0+net10.0 / unit+integration)

Information
Class: AsyncResponse.Channels.NATS.NatsConsumeLoopException
Assembly: AsyncResponse.Channels.NATS
File(s): /_/src/Channels/AsyncResponse.Channels.NATS/NatsAsyncResponseChannel.cs
Line coverage
100%
Covered lines: 1
Uncovered lines: 0
Coverable lines: 1
Total lines: 1035
Line coverage: 100%
Branch coverage
N/A
Covered branches: 0
Total branches: 0
Branch coverage: N/A
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%11100%

File(s)

/_/src/Channels/AsyncResponse.Channels.NATS/NatsAsyncResponseChannel.cs

#LineLine coverage
 1using Microsoft.Extensions.DependencyInjection;
 2using Microsoft.Extensions.Logging;
 3using Microsoft.Extensions.Options;
 4using System.Buffers;
 5using System.Diagnostics;
 6using System.Text;
 7using System.Text.Json;
 8
 9namespace AsyncResponse.Channels.NATS;
 10
 11/// <summary>
 12/// NATS-backed response channel:
 13/// <list type="bullet">
 14/// <item><description>Delivers responses over NATS Core request/reply on a subject keyed by
 15/// correlation id: a waiter subscribes and acks each message, and the publisher requests so the NATS
 16/// "no responders" signal reports precisely when nobody is listening.</description></item>
 17/// <item><description>Persists <see cref="RecoveryState"/> in a JetStream Key-Value bucket so a
 18/// response arriving after the waiter died (e.g. a redeploy) is routed through the lost-subscriber
 19/// dispatcher, which asks the payload's OnRecovery and invokes the resume or failure
 20/// callback with the materialized payload (or keeps the registration armed for a checkpoint).</description></item>
 21/// </list>
 22/// </summary>
 23internal sealed class NatsAsyncResponseChannel : IAsyncResponsePublisher, IRawAsyncResponsePublisher, IRecoverableAsyncR
 24{
 25    private readonly INatsResponseChannelClient _client;
 26    private readonly IRecoveryStateStore _recoveryStateStore;
 27    private readonly AsyncResponseContextPropagation _propagation;
 28    private readonly LostSubscriberCallbackDispatcher _lostSubscriberDispatcher;
 29    private readonly NatsSubjectSchema _subjects;
 30    private readonly NatsAsyncResponseChannelOptions _options;
 31    private readonly ILogger<NatsAsyncResponseChannel> _logger;
 32    private readonly TimeProvider _timeProvider;
 33
 34    /// <summary>Creates a NATS-backed async-response channel.</summary>
 35    public NatsAsyncResponseChannel(
 36        IServiceScopeFactory scopeFactory,
 37        INatsResponseChannelClient client,
 38        IRecoveryStateStore recoveryStateStore,
 39        IOptions<NatsAsyncResponseChannelOptions> options,
 40        AsyncResponseContextPropagation propagation,
 41        ILogger<NatsAsyncResponseChannel> logger,
 42        TimeProvider? timeProvider = null)
 43    {
 44        _timeProvider = timeProvider ?? TimeProvider.System;
 45        _options = options.Value;
 46        _options.Validate();
 47        _client = client;
 48        _recoveryStateStore = recoveryStateStore;
 49        _propagation = propagation;
 50        _subjects = new NatsSubjectSchema(_options.SubjectPrefix);
 51        _logger = logger;
 52        _lostSubscriberDispatcher = new LostSubscriberCallbackDispatcher(scopeFactory, propagation, logger, _timeProvide
 53    }
 54
 55    // ---------------------------------------------------------------------------------------
 56    // IAsyncResponseSubscriber / IRecoverableAsyncResponseSubscriber
 57
 58    /// <inheritdoc/>
 59    public Task<IAsyncResponseWaiter<T>> CreateResponseWaiter<T>(
 60        string correlationId,
 61        Func<T, ValueTask<bool>>? completionPredicate = null,
 62        TimeSpan? timeout = null) where T : IAsyncResponsePayload
 63        => CreateResponseWaiterCore(correlationId, resumeCallback: null, failureCallback: null, completionPredicate, tim
 64
 65    /// <inheritdoc/>
 66    public Task<IAsyncResponseWaiter<T>> CreateRecoverableResponseWaiter<T>(
 67        string correlationId,
 68        ReflectionCallDto? resumeCallback = null,
 69        ReflectionCallDto? failureCallback = null,
 70        Func<T, ValueTask<bool>>? completionPredicate = null,
 71        TimeSpan? timeout = null) where T : IAsyncResponsePayload
 72        => CreateResponseWaiterCore(correlationId, resumeCallback, failureCallback, completionPredicate, timeout);
 73
 74    /// <summary>
 75    /// The public <c>ResponseTask</c> surface: internal loop-fault settlements are marked with
 76    /// <see cref="NatsConsumeLoopException"/> so registration can abort atomically, but callers
 77    /// observe the original exception exactly as before.
 78    /// </summary>
 79    private static async Task<T> UnwrapConsumeLoopFaults<T>(Task<T> task) where T : IAsyncResponsePayload
 80    {
 81        try
 82        {
 83            return await task.ConfigureAwait(false);
 84        }
 85        catch (NatsConsumeLoopException loopFault)
 86        {
 87            System.Runtime.ExceptionServices.ExceptionDispatchInfo.Capture(loopFault.InnerException!).Throw();
 88            throw; // unreachable
 89        }
 90    }
 91
 92    private async Task<IAsyncResponseWaiter<T>> CreateResponseWaiterCore<T>(
 93        string correlationId,
 94        ReflectionCallDto? resumeCallback,
 95        ReflectionCallDto? failureCallback,
 96        Func<T, ValueTask<bool>>? completionPredicate,
 97        TimeSpan? timeout) where T : IAsyncResponsePayload
 98    {
 99        CorrelationIdGuard.ThrowIfUnusable(correlationId);
 100
 101        // Recovery callbacks only make sense if the payload can say whether a late response should
 102        // resume or fail the flow. On this durable channel that decision is real (it survives a
 103        // redeploy), so require the override rather than letting the conservative default silently
 104        // route every recovered response to the failure callback.
 105        if ((resumeCallback is not null || failureCallback is not null)
 106            && !AsyncResponsePayloadReflection.OverridesOnRecovery(typeof(T)))
 107        {
 108            throw new InvalidOperationException(
 109                $"Payload type '{typeof(T)}' registers lost-subscriber recovery callbacks on the NATS channel " +
 110                $"but does not override {nameof(IAsyncResponsePayload)}.{nameof(IAsyncResponsePayload.OnRecovery)}(). " 
 111                "Override it to declare what each response does to the flow — RecoveryAction.Resume, " +
 112                "RecoveryAction.Fail, or RecoveryAction.KeepWaiting for non-terminal checkpoints; the durable " +
 113                "channel needs this to route a response that arrives after the waiter was lost.");
 114        }
 115
 116        // default: first envelope completes the wait
 117        completionPredicate ??= _ => new ValueTask<bool>(true);
 118
 119        // Default timeout aligned with the recovery-state expiry: an infinite wait is never
 120        // meaningful, because once the recovery state expires the correlation id has no recovery
 121        // anyway. Timing out routes the flow through its normal failure handling instead of
 122        // leaving it stuck forever.
 123        timeout ??= _options.DefaultTimeout ?? _options.RecoveryStateExpiry;
 124        // BEFORE any side effect: an unsupported resolved timeout (non-positive, or past the
 125        // ~49.7-day BCL timer ceiling) used to throw only at timer arming — after the
 126        // subscription and recovery state existed, leaking both — and zero used to slip through
 127        // on some channels entirely, insta-timing-out a fully registered waiter.
 128        AsyncResponseChannelOptions.EnsureWaiterTimeoutSupported(timeout.Value);
 129
 130        var storedCorrelationId = correlationId;
 131        // Capture the subscribe-time ExecutionContext so app AsyncLocals (trace, principal, logging
 132        // scope) flow into the message handler, which runs on a background consume-loop thread.
 133        var capturedContext = ExecutionContext.Capture();
 134        var subject = _subjects.ResponseSubject(correlationId);
 135
 136        var activity = AsyncResponseDiagnostics.StartActivity("asyncresponse.wait", correlationId: correlationId);
 137        activity?.SetTag("asyncresponse.channel", "nats");
 138        AsyncResponseDiagnostics.SetPayloadType(activity, typeof(T));
 139        activity?.SetTag("asyncresponse.timeout_seconds", timeout.Value.TotalSeconds);
 140
 141        if (_logger.IsEnabled(LogLevel.Debug))
 142            _logger.LogDebug("Waiting for response on correlationId {CorrelationId} with timeout {Timeout}.", correlatio
 143
 144        var tcs = new TaskCompletionSource<T>(TaskCreationOptions.RunContinuationsAsynchronously);
 145        var registrationId = Guid.NewGuid();
 146
 147        // Single-use cancellation token implementing the timeout. Armed only after subscribe + recovery
 148        // save succeed, but its callback is registered first so a very fast terminal message cleans up safely.
 149        // Clock-injected (DbChannelShared parity): CancelAfter on a default CTS is bound to the
 150        // system clock, so a virtual clock could never fire a production-sized waiter timeout.
 151        var cancellationTokenSource = new CancellationTokenSource(Timeout.InfiniteTimeSpan, _timeProvider);
 152        CancellationTokenRegistration timeoutRegistration = default;
 153        INatsChannelSubscription? subscription = null;
 154
 155        // -------------------------------------------------------------------------
 156        // Local: CleanupOnceAsync — ends the stream (which ends the consume loop), deletes
 157        // recovery state, and tears down the timeout, exactly once.
 158        int cleanupStarted = 0;
 159        int teardownBudgetSpent = 0;
 160        var subscriptionTornDown = false;
 161        var cleanupGate = new object();
 162        Task? cleanupTask = null;
 163        var streamEndGate = new object();
 164        Task? streamEndTask = null;
 165        var consumeLoop = Task.CompletedTask;
 166
 167        // The ONE place the server-side subscription is disposed — the drain and the latched
 168        // cleanup both need the stream ended (whichever runs first), and having each dispose it
 169        // independently doubled the teardown for no benefit. TASK-latched and NEVER-faulting:
 170        // its failure is logged here exactly once, no matter how many latched callers observe
 171        // the task — and a caller that abandoned its bounded wait still gets the late outcome
 172        // recorded instead of it dying as a TaskScheduler.UnobservedTaskException. Callers read
 173        // "completed with subscriptionTornDown false" as teardown failure and backstop-cancel
 174        // (the cleanup core's finally, safe only after the timeout registration is gone).
 175        Task EndStreamOnce()
 176        {
 177            lock (streamEndGate)
 178            {
 179                return streamEndTask ??= EndStreamCoreAsync();
 180            }
 181        }
 182
 183        async Task EndStreamCoreAsync()
 184        {
 185            if (subscription is null)
 186                return;
 187
 188            try
 189            {
 190                await subscription.DisposeAsync().ConfigureAwait(false);
 191                subscriptionTornDown = true;
 192                _logger.LogDebug("Unsubscribed from subject {Subject}.", subject);
 193            }
 194            catch (Exception teardownEx)
 195            {
 196                _logger.LogError(teardownEx, "Error during cleanup for subject {Subject}.", subject);
 197            }
 198        }
 199
 200        // Task-latched so EVERY caller completes only when the one real cleanup has finished —
 201        // the previous fire-once int latch let a second caller (a disposing waiter racing the
 202        // timeout) return before the task was settled. The core itself never waits on the consume
 203        // loop: draining happens BEFORE the latch (DrainThenCleanupAsync), because the loop's own
 204        // finally also enters this latch — a join inside the core would make the loop await a core
 205        // that is joining the loop.
 206        ValueTask CleanupOnceAsync()
 207        {
 208            Task task;
 209            lock (cleanupGate)
 210            {
 211                task = cleanupTask ??= CleanupCoreAsync();
 212            }
 213
 214            return task.IsCompletedSuccessfully ? ValueTask.CompletedTask : new ValueTask(task);
 215        }
 216
 217        // Dispose-path cleanup: DRAINS the in-flight delivery before settling. The consume loop
 218        // may be mid Until-predicate holding a claimed terminal message; ending the stream and
 219        // joining the loop guarantees that by the time the latched core cancels, the task is
 220        // either settled by that delivery or genuinely undelivered. Never called from the loop
 221        // itself — loop-invoked cleanup uses CleanupOnceAsync directly, its task already settled
 222        // by the terminal dispatch.
 223        //
 224        // One DisposalDrainTimeout budget covers BOTH steps — a wedged client library can hang
 225        // the subscription dispose just as a wedged Until predicate can hang the loop join. The
 226        // core's cancel is only truthful once the JOIN below has proven the loop ended; any
 227        // drain outcome short of that — budget lapse, anything unforeseen — leaves a delivery
 228        // possibly mid-predicate holding a message already consumed from the stream, and
 229        // "canceled" would tell a re-attaching caller nothing was delivered. Those paths fault
 230        // the task with the explicit indeterminate contract instead (routing durable flows to a
 231        // fresh idempotent restart) and cancel the subscription token so the loop still ends
 232        // once the predicate returns. (A teardown FAILURE no longer throws — the latched
 233        // teardown logs it and leaves subscriptionTornDown false — so it backstop-cancels and
 234        // still proves settlement through the join.)
 235        async ValueTask DrainThenCleanupAsync(Exception? terminalIfUndelivered = null)
 236        {
 237            if (Volatile.Read(ref cleanupStarted) == 0)
 238            {
 239                var drainTimeout = _options.DisposalDrainTimeout;
 240                // ONE budget for the whole disposal: the latched cleanup below skips its own
 241                // teardown wait when a drain already spent this budget on the same latched
 242                // teardown task — a second full wait there made disposal cost double the
 243                // configured DisposalDrainTimeout.
 244                Volatile.Write(ref teardownBudgetSpent, 1);
 245                try
 246                {
 247                    using var budget = new CancellationTokenSource(drainTimeout);
 248                    await EndStreamOnce().WaitAsync(budget.Token).ConfigureAwait(false);
 249
 250                    // A failed teardown surfaces as "completed, subscriptionTornDown false" (the
 251                    // latched core logged it): backstop-cancel so the loop still ends, then FALL
 252                    // THROUGH to the join — a failed teardown proves nothing about a delivery
 253                    // mid-predicate, and skipping the join here let cleanup cancel a response the
 254                    // stream had already handed over.
 255                    if (!subscriptionTornDown && subscription is not null)
 256                        await DisarmThenCancelSubscriptionTokenAsync().ConfigureAwait(false);
 257
 258                    // Settlement is PROVEN only by the loop having ended within the remaining
 259                    // budget — either it settled the task with the in-flight delivery, or it
 260                    // ended with nothing in flight and the cleanup's cancel below is truthful.
 261                    await consumeLoop.WaitAsync(budget.Token).ConfigureAwait(false);
 262                }
 263                catch (Exception drainEx)
 264                {
 265                    _logger.LogWarning(
 266                        "Disposal drain for correlationId {CorrelationId} did not prove settlement within {DrainTimeout}
 267                        correlationId, drainTimeout);
 268                    AsyncResponseDiagnostics.SetError(activity, "indeterminate_delivery", "Disposal drain did not prove 
 269                    // A TrySetResult from the late-finishing delivery loses against this and is
 270                    // dropped; the loop's own cleanup call is a no-op behind the latch. The
 271                    // non-cancellation exception case is unforeseen infrastructure failure —
 272                    // settlement is equally unproven there, so it must not fall back to cancel.
 273                    if (drainEx is not OperationCanceledException)
 274                        _logger.LogDebug(drainEx, "Disposal drain failed for subject {Subject}.", subject);
 275                    tcs.TrySetException(new AsyncResponseIndeterminateDeliveryException(correlationId, drainTimeout));
 276                    await DisarmThenCancelSubscriptionTokenAsync().ConfigureAwait(false);
 277                }
 278            }
 279
 280            // Settle AFTER the drain, never before it. The consume loop may already hold a message
 281            // the subscription received — the publisher was told "delivered", so it exists nowhere
 282            // else. Faulting first let a timeout beat that in-flight delivery and report a consumed
 283            // response as a timeout; TrySet loses here if the delivery won, which is the whole
 284            // point. (A lapsed drain budget has already faulted the task as indeterminate above,
 285            // and TrySet is a no-op behind it.)
 286            if (terminalIfUndelivered is not null)
 287                tcs.TrySetException(terminalIfUndelivered);
 288
 289            await CleanupOnceAsync().ConfigureAwait(false);
 290        }
 291
 292        async ValueTask DisarmThenCancelSubscriptionTokenAsync()
 293        {
 294            // Disarm the waiter-timeout registration BEFORE the backstop cancel — the cancel
 295            // would otherwise fire it and stamp a spurious TimeoutException plus a waiter-timeout
 296            // metric onto a disposal that is not a timeout. Idempotent with the cleanup core's
 297            // own registration disposal.
 298            await timeoutRegistration.DisposeAsync().ConfigureAwait(false);
 299            try
 300            {
 301                cancellationTokenSource.Cancel();
 302            }
 303            catch (ObjectDisposedException)
 304            {
 305                // Cleanup already ran and disposed the source; the loop is ending regardless.
 306            }
 307        }
 308
 309        async Task CleanupCoreAsync()
 310        {
 311            Interlocked.Exchange(ref cleanupStarted, 1);
 312
 313            try
 314            {
 315                try
 316                {
 317                    // Delete the recovery state BEFORE disposing the subscription. In the reverse
 318                    // order a publish landing in the window sees "no responders, state present" and
 319                    // fires a spurious recovery callback for a wait that already reached a terminal
 320                    // state. In this order the window shows a subscriber that drops the message — a
 321                    // late or duplicate terminal message is droppable; a resurrected recovery callback
 322                    // is not.
 323                    await _recoveryStateStore.TryDeleteAsync(correlationId, registrationId).ConfigureAwait(false);
 324                }
 325                catch (Exception ex)
 326                {
 327                    // Best-effort: the KV entry expires on its own, and a transient store failure
 328                    // must not skip the subscription teardown below.
 329                    _logger.LogError(ex, "Failed to delete recovery state for correlationId {CorrelationId}.", correlati
 330                }
 331
 332                // End the stream if the drain has not already — dispatch-triggered cleanup (a
 333                // terminal delivery, a loop fault) reaches here without a drain. The task-latch
 334                // keeps the teardown single no matter which path got here first. Bounded like the
 335                // drain: this latched core is what a disposing waiter awaits when terminal
 336                // delivery started cleanup first (the drain skips itself on cleanupStarted), so
 337                // an unbudgeted teardown here let a wedged client library hold DisposeAsync
 338                // hostage past DisposalDrainTimeout. On the bound lapsing, the catch below logs
 339                // and the finally's backstop cancel still ends the consume loop; the abandoned
 340                // teardown task never faults (it logs its own late outcome). When a DRAIN
 341                // preceded this core, it already spent that budget on this same latched task —
 342                // waiting a second one here made disposal cost double the configured bound, so
 343                // the wait is skipped (the teardown is running and self-logging regardless).
 344                if (Volatile.Read(ref teardownBudgetSpent) == 0)
 345                    await EndStreamOnce().WaitAsync(_options.DisposalDrainTimeout).ConfigureAwait(false);
 346                if (subscription is null)
 347                    subscriptionTornDown = true;
 348            }
 349            catch (Exception ex)
 350            {
 351                _logger.LogError(ex, "Error during cleanup for subject {Subject}.", subject);
 352            }
 353            finally
 354            {
 355                await timeoutRegistration.DisposeAsync().ConfigureAwait(false);
 356                if (!subscriptionTornDown)
 357                {
 358                    // DisposeAsync did not complete, so the server-side subscription may still be
 359                    // pumping messages. Its lifetime is bound to this token (SubscribeAsync received
 360                    // it), and disposing a CTS never cancels — an explicit cancel is the backstop
 361                    // that ends the consume loop. Safe only after the timeout registration above is
 362                    // gone, or the cancel would fire a spurious waiter timeout.
 363                    cancellationTokenSource.Cancel();
 364                }
 365
 366                // A waiter disposed before any terminal signal must not leave ResponseTask pending
 367                // forever for callers that hold it directly — the timeout died above, so nothing
 368                // else could ever complete the task. A no-op after a normal completion, timeout,
 369                // fault, or a delivery drained by DrainThenCleanupAsync.
 370                tcs.TrySetCanceled();
 371
 372                cancellationTokenSource.Dispose();
 373                activity?.Dispose();
 374            }
 375        }
 376
 377        // -------------------------------------------------------------------------
 378        // Local: ProcessResponseAsync — deserializes and handles a single envelope, completes the TCS when terminal.
 379        async Task ProcessResponseAsync(string? payload)
 380        {
 381            bool finished = false;
 382            try
 383            {
 384                if (string.IsNullOrEmpty(payload))
 385                {
 386                    // A non-probe message with no body cannot be a response; ignore it rather than fault.
 387                    _logger.LogWarning("Received empty response message for correlationId {CorrelationId}; ignoring.", c
 388                    return;
 389                }
 390
 391                // JsonSafety, not the raw reader: a parse failure is logged below and handed to the
 392                // waiter, and the reader's own message quotes inbound property names and dictionary
 393                // keys (docs/security.md, "never logs a message body"). Size and position only.
 394                var envelope = JsonSafety.SafeDeserialize(payload, AsyncResponseEnvelopeJson.TypeInfo<T>());
 395
 396                if (envelope == null)
 397                {
 398                    _logger.LogError("Failed to deserialize envelope for correlationId {CorrelationId}.", correlationId)
 399                    finished = true;
 400                    var deserializationError = new JsonException($"Failed to deserialize envelope for correlationId {cor
 401                    AsyncResponseDiagnostics.SetError(activity, "deserialize_failure", deserializationError.Message);
 402                    if (!tcs.TrySetException(deserializationError))
 403                        _logger.LogWarning(deserializationError, "TaskCompletionSource already completed for correlation
 404                }
 405                else if (!AsyncResponseEnvelopeSchema.IsReadable(envelope.SchemaVersion))
 406                {
 407                    finished = true;
 408                    var schemaError = new InvalidOperationException(
 409                        $"Response envelope for correlationId {correlationId} has schema version {envelope.SchemaVersion
 410                        $"which this build does not support (current: {AsyncResponseEnvelopeSchema.Current}).");
 411                    AsyncResponseDiagnostics.SetError(activity, "schema_mismatch", schemaError.Message);
 412                    if (!tcs.TrySetException(schemaError))
 413                        _logger.LogWarning(schemaError, "TaskCompletionSource already completed for correlationId {Corre
 414                }
 415                else if (!envelope.Success)
 416                {
 417                    finished = true;
 418                    var remoteFailure = new Exception(envelope.ExceptionMessage ?? "Unknown error during asynchronous pr
 419                    if (!string.IsNullOrEmpty(envelope.ExceptionStackTrace))
 420                        // Cap on receive too: the publish-side cap only bounds traces we emit, not what
 421                        // a remote we do not control can push at us.
 422                        remoteFailure.Data["RemoteStackTrace"] = RemoteStackTrace.Cap(envelope.ExceptionStackTrace, _opt
 423
 424                    _logger.LogWarning("Received error response for correlationId {CorrelationId}: {ErrorMessage}", corr
 425                    AsyncResponseDiagnostics.SetError(activity, "remote_failure", remoteFailure.Message);
 426                    if (!tcs.TrySetException(remoteFailure))
 427                        _logger.LogWarning(remoteFailure, "TaskCompletionSource already completed for correlationId {Cor
 428                }
 429                else
 430                {
 431                    if (_logger.IsEnabled(LogLevel.Debug))
 432                        _logger.LogDebug("Received response for correlationId {CorrelationId}.", correlationId);
 433                    finished = await completionPredicate(envelope.Payload!).ConfigureAwait(false);
 434                    if (finished && !tcs.TrySetResult(envelope.Payload!))
 435                        _logger.LogWarning("TaskCompletionSource already completed for correlationId {CorrelationId}.", 
 436                }
 437            }
 438            catch (Exception ex)
 439            {
 440                _logger.LogError(ex, "Error processing message on subject {Subject} for correlationId {CorrelationId}.",
 441                finished = true;
 442                AsyncResponseDiagnostics.SetError(activity, ex);
 443                if (!tcs.TrySetException(ex))
 444                    _logger.LogWarning(ex, "TaskCompletionSource already completed for correlationId {CorrelationId}.", 
 445            }
 446            finally
 447            {
 448                if (finished)
 449                    await CleanupOnceAsync().ConfigureAwait(false);
 450            }
 451        }
 452
 453        // -------------------------------------------------------------------------
 454        // Local: ProcessUnderCapturedContextAsync — restores the waiter's subscribe-time
 455        // ExecutionContext (app AsyncLocals) plus the correlation id before processing, since the
 456        // consume loop runs on a background thread that never had them.
 457        Task ProcessUnderCapturedContextAsync(string? payload)
 458        {
 459            async Task Process()
 460            {
 461                using var correlationScope = AsyncResponseContext.PushCorrelationId(storedCorrelationId);
 462                await ProcessResponseAsync(payload).ConfigureAwait(false);
 463            }
 464
 465            if (capturedContext is null)
 466                return Process();
 467
 468            Task? task = null;
 469            ExecutionContext.Run(capturedContext, _ => task = Process(), null);
 470            return task!;
 471        }
 472
 473        // -------------------------------------------------------------------------
 474        // Local: ConsumeLoopAsync — reads messages serially from the subscription until it is disposed.
 475        async Task ConsumeLoopAsync(INatsChannelSubscription sub)
 476        {
 477            try
 478            {
 479                await foreach (var message in sub.ReadAsync(CancellationToken.None).ConfigureAwait(false))
 480                {
 481                    // Ack first so the publisher's request resolves quickly (delivery/liveness confirmed)
 482                    // even if processing the payload is slow. A failed ack must not abort the wait.
 483                    try
 484                    {
 485                        await message.ReplyAsync().ConfigureAwait(false);
 486                    }
 487                    catch (Exception replyEx)
 488                    {
 489                        _logger.LogDebug(replyEx, "Failed to acknowledge response on subject {Subject}.", subject);
 490                    }
 491
 492                    if (message.IsProbe)
 493                        continue;
 494
 495                    await ProcessUnderCapturedContextAsync(message.Payload).ConfigureAwait(false);
 496                }
 497            }
 498            catch (Exception ex)
 499            {
 500                _logger.LogError(ex, "Response subscription loop failed for subject {Subject}.", subject);
 501                AsyncResponseDiagnostics.SetError(activity, ex);
 502                // The settlement itself carries its source: a loop death is a TRANSPORT failure,
 503                // not a delivered response, and the registration path must tell the two kinds of
 504                // faulted task apart ATOMICALLY — a side-band flag raced the settlement in both
 505                // directions (set-then-lose aborted a registration whose response had already
 506                // been delivered; set-after-win left a window that returned a dead waiter). Only
 507                // a loop fault that actually WINS the settlement marks the task; a fault that
 508                // loses to a terminal payload changes nothing. The wrapper never escapes: the
 509                // public ResponseTask unwraps it back to the original exception.
 510                if (!tcs.TrySetException(new NatsConsumeLoopException(ex)))
 511                    _logger.LogWarning(ex, "TaskCompletionSource already completed for correlationId {CorrelationId}.", 
 512                await CleanupOnceAsync().ConfigureAwait(false);
 513            }
 514        }
 515
 516        timeoutRegistration = cancellationTokenSource.Token.Register(() =>
 517        {
 518            _ = Task.Run(async () =>
 519            {
 520                try
 521                {
 522                    _logger.LogWarning("Timed out waiting for response for correlationId {CorrelationId}.", correlationI
 523                    AsyncResponseDiagnostics.SetError(activity, "timeout", $"Timed out waiting for response for correlat
 524                    AsyncResponseDiagnostics.RecordWaiterTimeout("nats");
 525                    await DrainThenCleanupAsync(
 526                        new TimeoutException($"Timed out waiting for response for correlationId {correlationId}."))
 527                        .ConfigureAwait(false);
 528                }
 529                catch (Exception ex)
 530                {
 531                    // Fire-and-forget: nothing awaits this task, so an escaped fault would vanish.
 532                    _logger.LogError(ex, "Error handling waiter timeout for correlationId {CorrelationId}.", correlation
 533                }
 534            });
 535        });
 536
 537        try
 538        {
 539            subscription = await _client.SubscribeAsync(subject, cancellationTokenSource.Token).ConfigureAwait(false);
 540            consumeLoop = Task.Run(() => ConsumeLoopAsync(subscription));
 541
 542            // Round-trip to the server so the subscription is guaranteed registered BEFORE the
 543            // recovery state is saved (and before the caller's trigger publishes the remote
 544            // request — closing the subscribe/trigger race). The order is the DB channels'
 545            // invariant: "recovery state visible ⇒ subscription visible". Saved first, a publish
 546            // landing in the window found the registration, probed the not-yet-visible
 547            // subscription, and consumed a live waiter's recovery arm — the waiter then resumed
 548            // twice (recovery callback now, live delivery to its timeout). Skipped once cleanup
 549            // started: the wait already settled terminally, so there is no trigger race left to
 550            // close — and cleanup's teardown disposes the lifetime source this flush reads its
 551            // token from, so attempting it would throw for nothing.
 552            if (Volatile.Read(ref cleanupStarted) == 0)
 553                await _client.FlushAsync(cancellationTokenSource.Token).ConfigureAwait(false);
 554
 555            var recoveryState = new RecoveryState
 556            {
 557                RegistrationId = registrationId,
 558                ResumeCallback = resumeCallback,
 559                FailureCallback = failureCallback,
 560                CorrelationId = correlationId,
 561                PayloadTypeFullName = typeof(T).FullName,
 562                // The engine's clock, not the ambient one. The watchdog judges staleness as
 563                // "utcNow - RegisteredAtUtc" from whichever host scans, so an unsubstitutable
 564                // app-clock stamp made a skewed host's registrations either never age (skew ahead:
 565                // a genuinely stuck flow stays invisible and the health check stays green) or age
 566                // instantly (skew behind: healthy waits page the operator every scan). The DB
 567                // channels stamp the SERVER clock for exactly this reason; this at least puts the
 568                // stamp and the watchdog's "now" on one substitutable clock, and matches the
 569                // ExpiresAtUtc the recovery store writes for the same registration.
 570                RegisteredAtUtc = _timeProvider.GetUtcNow().UtcDateTime,
 571                Context = _propagation.Capture()
 572            };
 573            await _recoveryStateStore.SaveAsync(correlationId, recoveryState, _options.RecoveryStateExpiry).ConfigureAwa
 574            if (Volatile.Read(ref cleanupStarted) != 0)
 575            {
 576                // A terminal delivery on the already-running consume loop started cleanup while
 577                // this registration was still being written: cleanup's delete ran before the save
 578                // committed, so the save just orphaned a callback-armed registration that would
 579                // resurrect recovery for a wait that already reached a terminal state. Compensate
 580                // with a second delete (mirrors the in-memory channel's post-save check).
 581                // Best-effort: TTL and the watchdog back a failed delete.
 582                try
 583                {
 584                    await _recoveryStateStore.TryDeleteAsync(correlationId, registrationId).ConfigureAwait(false);
 585                }
 586                catch (Exception ex)
 587                {
 588                    _logger.LogError(ex, "Post-save recovery-state compensation delete failed for correlationId {Correla
 589                }
 590            }
 591
 592            _logger.LogDebug("Subscribed to subject {Subject} for correlationId {CorrelationId}.", subject, correlationI
 593        }
 594        catch (Exception ex) when (tcs.Task.IsCompletedSuccessfully
 595                                   || (tcs.Task.IsFaulted && tcs.Task.Exception!.InnerException is not NatsConsumeLoopEx
 596        {
 597            // The wait already settled: a delivery on the consume loop completed the waiter while
 598            // this registration step was still in flight (cleanup marks cleanupStarted just after
 599            // setting the task, so the task is the race-free signal), and the step then failed
 600            // against the torn-down registration state (a failed save, a flush aborted by the
 601            // disposed lifetime source). The response in hand outranks the builder's
 602            // "throw so the trigger never fires" contract — rethrowing would discard a delivered
 603            // response, the exact loss this library exists to prevent, and the success path for
 604            // this same interleaving already returns the completed waiter. Cleanup runs on the
 605            // delivery path, so nothing is leaked; a save that still committed is compensated
 606            // above or expires via TTL, with the recovery watchdog behind it. The filter demands
 607            // an actual settlement (result or fault): a canceled task means NO response was
 608            // delivered — e.g. a future channel-wide teardown canceling in-flight registrations —
 609            // and takes the rethrow path below.
 610            _logger.LogWarning(ex,
 611                "Registration step failed after a delivery settled correlationId {CorrelationId}; returning the complete
 612                correlationId);
 613        }
 614        catch (Exception ex)
 615        {
 616            _logger.LogError(ex, "Failed to subscribe to subject {Subject} for correlationId {CorrelationId}.", subject,
 617            AsyncResponseDiagnostics.SetError(activity, "subscribe_failure", ex.Message);
 618            await DrainThenCleanupAsync().ConfigureAwait(false);
 619
 620            // Rethrow instead of returning a pre-faulted waiter: the builder's contract is that
 621            // the trigger runs only once the subscription AND recovery state exist. A returned
 622            // waiter would still let the trigger fire the remote operation with no registration
 623            // left to receive (or recover) its response. Cleanup leaves nothing behind and cancels
 624            // the response task rather than faulting it, so no unobserved fault lingers.
 625            throw;
 626        }
 627
 628        if (tcs.Task.IsFaulted && tcs.Task.Exception!.InnerException is NatsConsumeLoopException loopFault)
 629        {
 630            // The consume loop died AND won the settlement while this registration was in
 631            // flight: the fault is a TRANSPORT error, not a delivered response; no subscription
 632            // is live; and the loop's cleanup already ran (deleting any saved recovery state,
 633            // backed by the post-save compensation). Returning the waiter would let the builder
 634            // fire the trigger with nothing registered to receive — or recover — its response,
 635            // so the builder contract applies: throw, and the remote operation never starts. A
 636            // loop fault that LOST the settlement leaves no mark, so a wait a terminal payload
 637            // already settled is returned normally.
 638            throw new InvalidOperationException(
 639                $"The NATS response subscription for correlationId {correlationId} failed before registration completed.
 640                loopFault.InnerException);
 641        }
 642
 643        try
 644        {
 645            if (Volatile.Read(ref cleanupStarted) == 0)
 646                cancellationTokenSource.CancelAfter(timeout.Value);
 647        }
 648        catch (ObjectDisposedException)
 649        {
 650            // A response completed and cleaned up between the check and CancelAfter.
 651        }
 652
 653        return new NatsAsyncResponseWaiter<T>(UnwrapConsumeLoopFaults(tcs.Task), () => DrainThenCleanupAsync());
 654    }
 655
 656    // ---------------------------------------------------------------------------------------
 657    // IAsyncResponsePublisher
 658
 659    /// <inheritdoc/>
 660    public Task SetResponse<T>(T response, string correlationId, CancellationToken cancellationToken = default) where T 
 661        => SetResponseCore(response, correlationId, cancellationToken);
 662
 663    Task IRawAsyncResponsePublisher.SetRawResponse(object? response, string correlationId, CancellationToken cancellatio
 664        => SetResponseCore(response, correlationId, cancellationToken);
 665
 666    Task IRawAsyncResponsePublisher.SetRawResponseJson(string responseJson, string correlationId, CancellationToken canc
 667        => SetRawResponseJsonCore(responseJson, correlationId, cancellationToken);
 668
 669    private async Task SetResponseCore<T>(T response, string correlationId, CancellationToken cancellationToken)
 670    {
 671        using var activity = AsyncResponseDiagnostics.StartActivity("asyncresponse.set_response", ActivityKind.Producer)
 672        activity?.SetTag("asyncresponse.channel", "nats");
 673        AsyncResponseDiagnostics.SetPayloadType(activity, typeof(T));
 674
 675        AsyncResponseDiagnostics.SetCorrelationId(activity, correlationId);
 676
 677        if (CorrelationIdGuard.IsUnpublishable(correlationId, _logger, activity, "the response"))
 678            return;
 679
 680        var subject = _subjects.ResponseSubject(correlationId);
 681        try
 682        {
 683            var envelope = new AsyncResponseEnvelope<T> { Success = true, Payload = response };
 684            var json = AsyncResponseEnvelopeJson.Serialize(envelope);
 685            var outcome = await _client.RequestAsync(subject, json, probe: false, _options.DeliveryConfirmationTimeout, 
 686            activity?.SetTag("asyncresponse.delivery", outcome.ToString());
 687
 688            if (outcome == NatsDeliveryOutcome.NoResponders)
 689            {
 690                // Nobody was listening (the waiter died, e.g. with a redeploy): hand the response over
 691                // to the lost-subscriber dispatcher, which asks the payload whether to resume or fail.
 692                var dispatchResult = await _lostSubscriberDispatcher
 693                    .DispatchLostResponses(
 694                        _recoveryStateStore,
 695                        correlationId,
 696                        response,
 697                        subject,
 698                        cancellationToken,
 699                        hasLiveSubscriber: () => HasLiveSubscriberAsync(correlationId, cancellationToken))
 700                    .ConfigureAwait(false);
 701                if (dispatchResult.RetryLive)
 702                {
 703                    // A waiter subscribed between the request and the recovery-state read —
 704                    // re-attempt the live publish instead of consuming its registration; only a
 705                    // second no-responders consumes it.
 706                    outcome = await _client.RequestAsync(subject, json, probe: false, _options.DeliveryConfirmationTimeo
 707                    activity?.SetTag("asyncresponse.delivery", outcome.ToString());
 708                    if (outcome != NatsDeliveryOutcome.NoResponders)
 709                        return;
 710
 711                    dispatchResult = await _lostSubscriberDispatcher
 712                        .DispatchLostResponses(
 713                            _recoveryStateStore,
 714                            correlationId,
 715                            response,
 716                            subject,
 717                            cancellationToken,
 718                            hasLiveSubscriber: () => HasLiveSubscriberAsync(correlationId, cancellationToken))
 719                        .ConfigureAwait(false);
 720                    if (dispatchResult.RetryLive)
 721                    {
 722                        // Second contradiction: delivery keeps reporting no responders while the
 723                        // probe keeps reporting a live subscriber (interest not yet visible
 724                        // server-side, or a stale heartbeat). Consuming registrations on this
 725                        // evidence would strip a live waiter of its recovery arm — leave all state
 726                        // intact and surface the non-delivery to the caller, whose retry/redelivery
 727                        // machinery re-attempts once the subscription is visible (bounded by the
 728                        // heartbeat's liveness expiry, after which normal recovery takes over).
 729                        // Returning here instead would silently drop the payload: the caller
 730                        // reports success, the broker message is acked, and the response then
 731                        // exists nowhere.
 732                        _logger.LogWarning(
 733                            "Delivery for correlationId {CorrelationId} found no subscribers twice while the liveness pr
 734                            correlationId);
 735                        activity?.SetTag("asyncresponse.recovery.liveness_contradiction", true);
 736                        throw new InvalidOperationException(
 737                            $"NATS delivery for correlationId '{correlationId}' found no responders twice while the live
 738                            "reporting a live subscriber; the payload was not delivered and recovery registrations were 
 739                            "the publish once the waiter's subscription is visible to the publishing endpoint.");
 740                    }
 741                }
 742
 743                AsyncResponseDiagnostics.SetLostSubscriberRoute(activity, dispatchResult.Action, dispatchResult.RouteMix
 744                AsyncResponseDiagnostics.RecordLostSubscriber("response", dispatchResult.Action, dispatchResult.Callback
 745                activity?.SetTag("asyncresponse.recovery.callback_invoked", dispatchResult.CallbackInvoked);
 746            }
 747            else if (_logger.IsEnabled(LogLevel.Debug))
 748            {
 749                _logger.LogDebug("Published response for correlationId {CorrelationId} on subject {Subject}. PayloadType
 750            }
 751        }
 752        catch (Exception ex)
 753        {
 754            _logger.LogError(ex, "Failed to publish response for correlationId {CorrelationId} on subject {Subject}.", c
 755            AsyncResponseDiagnostics.SetError(activity, ex);
 756            throw;
 757        }
 758    }
 759
 760    private async Task SetRawResponseJsonCore(string responseJson, string correlationId, CancellationToken cancellationT
 761    {
 762        using var activity = AsyncResponseDiagnostics.StartActivity("asyncresponse.ingress.raw_response", ActivityKind.P
 763        activity?.SetTag("asyncresponse.channel", "nats");
 764
 765        AsyncResponseDiagnostics.SetCorrelationId(activity, correlationId);
 766
 767        if (CorrelationIdGuard.IsUnpublishable(correlationId, _logger, activity, "the raw response", dropContractViolati
 768            return;
 769
 770        var subject = _subjects.ResponseSubject(correlationId);
 771        try
 772        {
 773            var json = SerializeRawSuccessEnvelope(responseJson);
 774            var outcome = await _client.RequestAsync(subject, json, probe: false, _options.DeliveryConfirmationTimeout, 
 775            activity?.SetTag("asyncresponse.delivery", outcome.ToString());
 776
 777            if (outcome == NatsDeliveryOutcome.NoResponders)
 778            {
 779                var response = new RawJsonResponse(responseJson).DeserializeUntyped();
 780
 781                var dispatchResult = await _lostSubscriberDispatcher
 782                    .DispatchLostResponses(
 783                        _recoveryStateStore,
 784                        correlationId,
 785                        response,
 786                        subject,
 787                        cancellationToken,
 788                        hasLiveSubscriber: () => HasLiveSubscriberAsync(correlationId, cancellationToken))
 789                    .ConfigureAwait(false);
 790                if (dispatchResult.RetryLive)
 791                {
 792                    // A waiter subscribed between the request and the recovery-state read —
 793                    // re-attempt the live publish instead of consuming its registration; only a
 794                    // second no-responders consumes it.
 795                    outcome = await _client.RequestAsync(subject, json, probe: false, _options.DeliveryConfirmationTimeo
 796                    activity?.SetTag("asyncresponse.delivery", outcome.ToString());
 797                    if (outcome != NatsDeliveryOutcome.NoResponders)
 798                        return;
 799
 800                    dispatchResult = await _lostSubscriberDispatcher
 801                        .DispatchLostResponses(
 802                            _recoveryStateStore,
 803                            correlationId,
 804                            response,
 805                            subject,
 806                            cancellationToken,
 807                            hasLiveSubscriber: () => HasLiveSubscriberAsync(correlationId, cancellationToken))
 808                        .ConfigureAwait(false);
 809                    if (dispatchResult.RetryLive)
 810                    {
 811                        // Second contradiction: delivery keeps reporting no responders while the
 812                        // probe keeps reporting a live subscriber (interest not yet visible
 813                        // server-side, or a stale heartbeat). Consuming registrations on this
 814                        // evidence would strip a live waiter of its recovery arm — leave all state
 815                        // intact and surface the non-delivery to the caller, whose retry/redelivery
 816                        // machinery re-attempts once the subscription is visible (bounded by the
 817                        // heartbeat's liveness expiry, after which normal recovery takes over).
 818                        // Returning here instead would silently drop the payload: the caller
 819                        // reports success, the broker message is acked, and the response then
 820                        // exists nowhere.
 821                        _logger.LogWarning(
 822                            "Delivery for correlationId {CorrelationId} found no subscribers twice while the liveness pr
 823                            correlationId);
 824                        activity?.SetTag("asyncresponse.recovery.liveness_contradiction", true);
 825                        throw new InvalidOperationException(
 826                            $"NATS delivery for correlationId '{correlationId}' found no responders twice while the live
 827                            "reporting a live subscriber; the payload was not delivered and recovery registrations were 
 828                            "the publish once the waiter's subscription is visible to the publishing endpoint.");
 829                    }
 830                }
 831
 832                AsyncResponseDiagnostics.SetLostSubscriberRoute(activity, dispatchResult.Action, dispatchResult.RouteMix
 833                AsyncResponseDiagnostics.RecordLostSubscriber("response", dispatchResult.Action, dispatchResult.Callback
 834                activity?.SetTag("asyncresponse.recovery.callback_invoked", dispatchResult.CallbackInvoked);
 835            }
 836            else if (_logger.IsEnabled(LogLevel.Debug))
 837            {
 838                _logger.LogDebug("Published raw response for correlationId {CorrelationId} on subject {Subject}. Outcome
 839            }
 840        }
 841        catch (Exception ex)
 842        {
 843            _logger.LogError(ex, "Failed to publish raw response for correlationId {CorrelationId} on subject {Subject}.
 844            AsyncResponseDiagnostics.SetError(activity, ex);
 845            throw;
 846        }
 847    }
 848
 849    /// <inheritdoc/>
 850    public async Task SetException(Exception exception, string correlationId, CancellationToken cancellationToken = defa
 851    {
 852        ArgumentNullException.ThrowIfNull(exception);
 853
 854        using var activity = AsyncResponseDiagnostics.StartActivity("asyncresponse.set_exception", ActivityKind.Producer
 855        activity?.SetTag("asyncresponse.channel", "nats");
 856        activity?.SetTag("asyncresponse.exception_type", exception.GetType().FullName ?? exception.GetType().Name);
 857
 858        AsyncResponseDiagnostics.SetCorrelationId(activity, correlationId);
 859
 860        if (CorrelationIdGuard.IsUnpublishable(correlationId, _logger, activity, "the exception", exception))
 861            return;
 862
 863        var subject = _subjects.ResponseSubject(correlationId);
 864        try
 865        {
 866            var envelope = new AsyncResponseEnvelope<object>
 867            {
 868                Success = false,
 869                ExceptionMessage = exception.Message,
 870                ExceptionStackTrace = RemoteStackTrace.ForWire(exception.StackTrace, _options.IncludeRemoteStackTrace, _
 871                Payload = null
 872            };
 873            var json = AsyncResponseEnvelopeJson.Serialize(envelope);
 874            var outcome = await _client.RequestAsync(subject, json, probe: false, _options.DeliveryConfirmationTimeout, 
 875            activity?.SetTag("asyncresponse.delivery", outcome.ToString());
 876
 877            if (outcome == NatsDeliveryOutcome.NoResponders)
 878            {
 879                // Nobody was listening: exception envelopes always go to the failure callback.
 880                var dispatchResult = await _lostSubscriberDispatcher
 881                    .DispatchLostExceptions(
 882                        _recoveryStateStore,
 883                        correlationId,
 884                        exception,
 885                        subject,
 886                        cancellationToken,
 887                        hasLiveSubscriber: () => HasLiveSubscriberAsync(correlationId, cancellationToken))
 888                    .ConfigureAwait(false);
 889                if (dispatchResult.RetryLive)
 890                {
 891                    // A waiter subscribed between the request and the recovery-state read —
 892                    // re-attempt the live publish instead of consuming its registration; only a
 893                    // second no-responders consumes it.
 894                    outcome = await _client.RequestAsync(subject, json, probe: false, _options.DeliveryConfirmationTimeo
 895                    activity?.SetTag("asyncresponse.delivery", outcome.ToString());
 896                    if (outcome != NatsDeliveryOutcome.NoResponders)
 897                        return;
 898
 899                    dispatchResult = await _lostSubscriberDispatcher
 900                        .DispatchLostExceptions(
 901                            _recoveryStateStore,
 902                            correlationId,
 903                            exception,
 904                            subject,
 905                            cancellationToken,
 906                            hasLiveSubscriber: () => HasLiveSubscriberAsync(correlationId, cancellationToken))
 907                        .ConfigureAwait(false);
 908                    if (dispatchResult.RetryLive)
 909                    {
 910                        // Second contradiction: delivery keeps reporting no responders while the
 911                        // probe keeps reporting a live subscriber (interest not yet visible
 912                        // server-side, or a stale heartbeat). Consuming registrations on this
 913                        // evidence would strip a live waiter of its recovery arm — leave all state
 914                        // intact and surface the non-delivery to the caller, whose retry/redelivery
 915                        // machinery re-attempts once the subscription is visible (bounded by the
 916                        // heartbeat's liveness expiry, after which normal recovery takes over).
 917                        // Returning here instead would silently drop the payload: the caller
 918                        // reports success, the broker message is acked, and the response then
 919                        // exists nowhere.
 920                        _logger.LogWarning(
 921                            "Delivery for correlationId {CorrelationId} found no subscribers twice while the liveness pr
 922                            correlationId);
 923                        activity?.SetTag("asyncresponse.recovery.liveness_contradiction", true);
 924                        throw new InvalidOperationException(
 925                            $"NATS delivery for correlationId '{correlationId}' found no responders twice while the live
 926                            "reporting a live subscriber; the payload was not delivered and recovery registrations were 
 927                            "the publish once the waiter's subscription is visible to the publishing endpoint.");
 928                    }
 929                }
 930
 931                activity?.SetTag("asyncresponse.recovery.callback_invoked", dispatchResult.CallbackInvoked);
 932                AsyncResponseDiagnostics.RecordLostSubscriber("exception", action: RecoveryAction.Fail, dispatchResult.C
 933            }
 934            else if (_logger.IsEnabled(LogLevel.Debug))
 935            {
 936                _logger.LogDebug("Published exception response for correlationId {CorrelationId} on subject {Subject}. O
 937            }
 938        }
 939        catch (Exception ex)
 940        {
 941            _logger.LogError(ex, "Failed to publish exception response for correlationId {CorrelationId} on subject {Sub
 942            AsyncResponseDiagnostics.SetError(activity, ex);
 943            throw;
 944        }
 945    }
 946
 947    // ---------------------------------------------------------------------------------------
 948    // IActiveSubscriberProbe
 949
 950    /// <inheritdoc/>
 951    public async ValueTask<long> CountActiveSubscribersAsync(string correlationId, CancellationToken cancellationToken =
 952    {
 953        if (string.IsNullOrWhiteSpace(correlationId))
 954            return 0L;
 955
 956        var subject = _subjects.ResponseSubject(correlationId);
 957        try
 958        {
 959            // NATS Core does not expose exact subscriber counts to clients, so the probe reports
 960            // presence. Only NoResponders is a definitive zero — the server told us nothing is
 961            // subscribed. NoReply means the OPPOSITE: interest existed and the ping was delivered,
 962            // it just was not acked inside PresenceProbeTimeout. That is routine for a LIVE waiter,
 963            // because the consume loop acks a probe only when it reads it, serially, after the
 964            // previous message's user Until predicate returns — so any predicate slower than the
 965            // 2s default made a healthy waiter look dead, which flagged it stale in the watchdog
 966            // and (worse) let the lost-subscriber dispatcher consume its recovery registration.
 967            var outcome = await _client.RequestAsync(subject, payload: null, probe: true, _options.PresenceProbeTimeout,
 968            return outcome switch
 969            {
 970                NatsDeliveryOutcome.Replied => 1L,
 971                NatsDeliveryOutcome.NoResponders => 0L,
 972                _ => -1L
 973            };
 974        }
 975        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
 976        {
 977            throw;
 978        }
 979        catch (Exception ex)
 980        {
 981            _logger.LogDebug(ex, "Failed to probe active subscribers for subject {Subject}.", subject);
 982            // Negative = "could not be probed" (the watchdog's unknown-liveness contract): 0 would
 983            // assert there is definitively no live waiter and flag every over-threshold
 984            // registration stale during a transient probe outage.
 985            return -1L;
 986        }
 987    }
 988
 989    /// <summary>
 990    /// Re-probes waiter liveness for the lost-subscriber dispatcher's snapshot-race re-check,
 991    /// using the same presence probe the watchdog uses. An unprobeable result THROWS instead of
 992    /// reading as "no live waiter", so the failure propagates to the publisher's catch and the
 993    /// publish retries rather than consuming a live waiter's recovery registration (parity with
 994    /// the DB channels, whose re-check calls the store directly).
 995    /// </summary>
 996    private async ValueTask<bool> HasLiveSubscriberAsync(string correlationId, CancellationToken cancellationToken)
 997    {
 998        var subscribers = await CountActiveSubscribersAsync(correlationId, cancellationToken).ConfigureAwait(false);
 999        if (subscribers < 0)
 1000        {
 1001            throw new InvalidOperationException(
 1002                $"NATS subscriber liveness for correlationId '{correlationId}' could not be probed.");
 1003        }
 1004
 1005        return subscribers > 0;
 1006    }
 1007
 1008    private static string SerializeRawSuccessEnvelope(string payloadJson)
 1009    {
 1010        JsonSafety.ThrowIfClearlyNotJson(payloadJson);
 1011
 1012        var buffer = new ArrayBufferWriter<byte>();
 1013        using (var writer = new Utf8JsonWriter(buffer))
 1014        {
 1015            writer.WriteStartObject();
 1016            writer.WriteNumber("SchemaVersion", AsyncResponseEnvelopeSchema.Current);
 1017            writer.WriteBoolean("Success", true);
 1018            writer.WritePropertyName("Payload");
 1019            writer.WriteRawValue(payloadJson);
 1020            writer.WriteNull("ExceptionMessage");
 1021            writer.WriteNull("ExceptionStackTrace");
 1022            writer.WriteEndObject();
 1023        }
 1024
 1025        return Encoding.UTF8.GetString(buffer.WrittenSpan);
 1026    }
 1027}
 1028
 1029/// <summary>
 1030/// Internal settlement marker: the consume loop faulted the wait (a transport failure — nothing
 1031/// was delivered). Never escapes the channel: registration converts a marked settlement into a
 1032/// thrown registration failure, and the public ResponseTask unwraps it to the original exception.
 1033/// </summary>
 1034internal sealed class NatsConsumeLoopException(Exception inner)
 81035    : Exception("The NATS response subscription loop failed.", inner);
Generated by: ReportGenerator 5.5.11.0
09/22/2026 - 09:41:04
GitHub | reportgenerator.io

Methods/Properties

.ctor(System.Exception)