| | | 1 | | using Microsoft.Extensions.DependencyInjection; |
| | | 2 | | using Microsoft.Extensions.Logging; |
| | | 3 | | using Microsoft.Extensions.Options; |
| | | 4 | | using System.Buffers; |
| | | 5 | | using System.Diagnostics; |
| | | 6 | | using System.Text; |
| | | 7 | | using System.Text.Json; |
| | | 8 | | |
| | | 9 | | namespace 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> |
| | | 23 | | internal 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> |
| | 505 | 35 | | public NatsAsyncResponseChannel( |
| | 505 | 36 | | IServiceScopeFactory scopeFactory, |
| | 505 | 37 | | INatsResponseChannelClient client, |
| | 505 | 38 | | IRecoveryStateStore recoveryStateStore, |
| | 505 | 39 | | IOptions<NatsAsyncResponseChannelOptions> options, |
| | 505 | 40 | | AsyncResponseContextPropagation propagation, |
| | 505 | 41 | | ILogger<NatsAsyncResponseChannel> logger, |
| | 505 | 42 | | TimeProvider? timeProvider = null) |
| | | 43 | | { |
| | 505 | 44 | | _timeProvider = timeProvider ?? TimeProvider.System; |
| | 505 | 45 | | _options = options.Value; |
| | 505 | 46 | | _options.Validate(); |
| | 505 | 47 | | _client = client; |
| | 505 | 48 | | _recoveryStateStore = recoveryStateStore; |
| | 505 | 49 | | _propagation = propagation; |
| | 505 | 50 | | _subjects = new NatsSubjectSchema(_options.SubjectPrefix); |
| | 505 | 51 | | _logger = logger; |
| | 505 | 52 | | _lostSubscriberDispatcher = new LostSubscriberCallbackDispatcher(scopeFactory, propagation, logger, _timeProvide |
| | 505 | 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 |
| | 235 | 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 |
| | 226 | 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 | | { |
| | 445 | 83 | | return await task.ConfigureAwait(false); |
| | | 84 | | } |
| | | 85 | | catch (NatsConsumeLoopException loopFault) |
| | | 86 | | { |
| | 2 | 87 | | System.Runtime.ExceptionServices.ExceptionDispatchInfo.Capture(loopFault.InnerException!).Throw(); |
| | 0 | 88 | | throw; // unreachable |
| | | 89 | | } |
| | 389 | 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 | | { |
| | 461 | 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. |
| | 455 | 105 | | if ((resumeCallback is not null || failureCallback is not null) |
| | 455 | 106 | | && !AsyncResponsePayloadReflection.OverridesOnRecovery(typeof(T))) |
| | | 107 | | { |
| | 2 | 108 | | throw new InvalidOperationException( |
| | 2 | 109 | | $"Payload type '{typeof(T)}' registers lost-subscriber recovery callbacks on the NATS channel " + |
| | 2 | 110 | | $"but does not override {nameof(IAsyncResponsePayload)}.{nameof(IAsyncResponsePayload.OnRecovery)}(). " |
| | 2 | 111 | | "Override it to declare what each response does to the flow — RecoveryAction.Resume, " + |
| | 2 | 112 | | "RecoveryAction.Fail, or RecoveryAction.KeepWaiting for non-terminal checkpoints; the durable " + |
| | 2 | 113 | | "channel needs this to route a response that arrives after the waiter was lost."); |
| | | 114 | | } |
| | | 115 | | |
| | | 116 | | // default: first envelope completes the wait |
| | 615 | 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. |
| | 453 | 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. |
| | 453 | 128 | | AsyncResponseChannelOptions.EnsureWaiterTimeoutSupported(timeout.Value); |
| | | 129 | | |
| | 451 | 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. |
| | 451 | 133 | | var capturedContext = ExecutionContext.Capture(); |
| | 451 | 134 | | var subject = _subjects.ResponseSubject(correlationId); |
| | | 135 | | |
| | 451 | 136 | | var activity = AsyncResponseDiagnostics.StartActivity("asyncresponse.wait", correlationId: correlationId); |
| | 451 | 137 | | activity?.SetTag("asyncresponse.channel", "nats"); |
| | 451 | 138 | | AsyncResponseDiagnostics.SetPayloadType(activity, typeof(T)); |
| | 451 | 139 | | activity?.SetTag("asyncresponse.timeout_seconds", timeout.Value.TotalSeconds); |
| | | 140 | | |
| | 451 | 141 | | if (_logger.IsEnabled(LogLevel.Debug)) |
| | 86 | 142 | | _logger.LogDebug("Waiting for response on correlationId {CorrelationId} with timeout {Timeout}.", correlatio |
| | | 143 | | |
| | 451 | 144 | | var tcs = new TaskCompletionSource<T>(TaskCreationOptions.RunContinuationsAsynchronously); |
| | 451 | 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. |
| | 451 | 151 | | var cancellationTokenSource = new CancellationTokenSource(Timeout.InfiniteTimeSpan, _timeProvider); |
| | 451 | 152 | | CancellationTokenRegistration timeoutRegistration = default; |
| | 451 | 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. |
| | 451 | 158 | | int cleanupStarted = 0; |
| | 451 | 159 | | int teardownBudgetSpent = 0; |
| | 451 | 160 | | var subscriptionTornDown = false; |
| | 451 | 161 | | var cleanupGate = new object(); |
| | 451 | 162 | | Task? cleanupTask = null; |
| | 451 | 163 | | var streamEndGate = new object(); |
| | 451 | 164 | | Task? streamEndTask = null; |
| | 451 | 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 | | { |
| | 451 | 177 | | lock (streamEndGate) |
| | | 178 | | { |
| | 451 | 179 | | return streamEndTask ??= EndStreamCoreAsync(); |
| | | 180 | | } |
| | 451 | 181 | | } |
| | | 182 | | |
| | | 183 | | async Task EndStreamCoreAsync() |
| | | 184 | | { |
| | 451 | 185 | | if (subscription is null) |
| | 0 | 186 | | return; |
| | | 187 | | |
| | | 188 | | try |
| | | 189 | | { |
| | 451 | 190 | | await subscription.DisposeAsync().ConfigureAwait(false); |
| | 445 | 191 | | subscriptionTornDown = true; |
| | 445 | 192 | | _logger.LogDebug("Unsubscribed from subject {Subject}.", subject); |
| | 445 | 193 | | } |
| | 6 | 194 | | catch (Exception teardownEx) |
| | | 195 | | { |
| | 6 | 196 | | _logger.LogError(teardownEx, "Error during cleanup for subject {Subject}.", subject); |
| | 6 | 197 | | } |
| | 451 | 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; |
| | 892 | 209 | | lock (cleanupGate) |
| | | 210 | | { |
| | 892 | 211 | | task = cleanupTask ??= CleanupCoreAsync(); |
| | 892 | 212 | | } |
| | | 213 | | |
| | 892 | 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 | | { |
| | 457 | 237 | | if (Volatile.Read(ref cleanupStarted) == 0) |
| | | 238 | | { |
| | 38 | 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. |
| | 38 | 244 | | Volatile.Write(ref teardownBudgetSpent, 1); |
| | | 245 | | try |
| | | 246 | | { |
| | 38 | 247 | | using var budget = new CancellationTokenSource(drainTimeout); |
| | 38 | 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. |
| | 36 | 255 | | if (!subscriptionTornDown && subscription is not null) |
| | 4 | 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. |
| | 36 | 261 | | await consumeLoop.WaitAsync(budget.Token).ConfigureAwait(false); |
| | 31 | 262 | | } |
| | 7 | 263 | | catch (Exception drainEx) |
| | | 264 | | { |
| | 7 | 265 | | _logger.LogWarning( |
| | 7 | 266 | | "Disposal drain for correlationId {CorrelationId} did not prove settlement within {DrainTimeout} |
| | 7 | 267 | | correlationId, drainTimeout); |
| | 7 | 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. |
| | 7 | 273 | | if (drainEx is not OperationCanceledException) |
| | 0 | 274 | | _logger.LogDebug(drainEx, "Disposal drain failed for subject {Subject}.", subject); |
| | 7 | 275 | | tcs.TrySetException(new AsyncResponseIndeterminateDeliveryException(correlationId, drainTimeout)); |
| | 7 | 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.) |
| | 457 | 286 | | if (terminalIfUndelivered is not null) |
| | 8 | 287 | | tcs.TrySetException(terminalIfUndelivered); |
| | | 288 | | |
| | 457 | 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. |
| | 11 | 298 | | await timeoutRegistration.DisposeAsync().ConfigureAwait(false); |
| | | 299 | | try |
| | | 300 | | { |
| | 11 | 301 | | cancellationTokenSource.Cancel(); |
| | 11 | 302 | | } |
| | 0 | 303 | | catch (ObjectDisposedException) |
| | | 304 | | { |
| | | 305 | | // Cleanup already ran and disposed the source; the loop is ending regardless. |
| | 0 | 306 | | } |
| | 11 | 307 | | } |
| | | 308 | | |
| | | 309 | | async Task CleanupCoreAsync() |
| | | 310 | | { |
| | 451 | 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. |
| | 451 | 323 | | await _recoveryStateStore.TryDeleteAsync(correlationId, registrationId).ConfigureAwait(false); |
| | 447 | 324 | | } |
| | 4 | 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. |
| | 4 | 329 | | _logger.LogError(ex, "Failed to delete recovery state for correlationId {CorrelationId}.", correlati |
| | 4 | 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). |
| | 451 | 344 | | if (Volatile.Read(ref teardownBudgetSpent) == 0) |
| | 413 | 345 | | await EndStreamOnce().WaitAsync(_options.DisposalDrainTimeout).ConfigureAwait(false); |
| | 449 | 346 | | if (subscription is null) |
| | 0 | 347 | | subscriptionTornDown = true; |
| | 449 | 348 | | } |
| | 2 | 349 | | catch (Exception ex) |
| | | 350 | | { |
| | 2 | 351 | | _logger.LogError(ex, "Error during cleanup for subject {Subject}.", subject); |
| | 2 | 352 | | } |
| | | 353 | | finally |
| | | 354 | | { |
| | 451 | 355 | | await timeoutRegistration.DisposeAsync().ConfigureAwait(false); |
| | 451 | 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. |
| | 8 | 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. |
| | 451 | 370 | | tcs.TrySetCanceled(); |
| | | 371 | | |
| | 451 | 372 | | cancellationTokenSource.Dispose(); |
| | 451 | 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 | | { |
| | 548 | 381 | | bool finished = false; |
| | | 382 | | try |
| | | 383 | | { |
| | 548 | 384 | | if (string.IsNullOrEmpty(payload)) |
| | | 385 | | { |
| | | 386 | | // A non-probe message with no body cannot be a response; ignore it rather than fault. |
| | 2 | 387 | | _logger.LogWarning("Received empty response message for correlationId {CorrelationId}; ignoring.", c |
| | 2 | 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. |
| | 546 | 394 | | var envelope = JsonSafety.SafeDeserialize(payload, AsyncResponseEnvelopeJson.TypeInfo<T>()); |
| | | 395 | | |
| | 538 | 396 | | if (envelope == null) |
| | | 397 | | { |
| | 6 | 398 | | _logger.LogError("Failed to deserialize envelope for correlationId {CorrelationId}.", correlationId) |
| | 6 | 399 | | finished = true; |
| | 6 | 400 | | var deserializationError = new JsonException($"Failed to deserialize envelope for correlationId {cor |
| | 6 | 401 | | AsyncResponseDiagnostics.SetError(activity, "deserialize_failure", deserializationError.Message); |
| | 6 | 402 | | if (!tcs.TrySetException(deserializationError)) |
| | 2 | 403 | | _logger.LogWarning(deserializationError, "TaskCompletionSource already completed for correlation |
| | | 404 | | } |
| | 532 | 405 | | else if (!AsyncResponseEnvelopeSchema.IsReadable(envelope.SchemaVersion)) |
| | | 406 | | { |
| | 6 | 407 | | finished = true; |
| | 6 | 408 | | var schemaError = new InvalidOperationException( |
| | 6 | 409 | | $"Response envelope for correlationId {correlationId} has schema version {envelope.SchemaVersion |
| | 6 | 410 | | $"which this build does not support (current: {AsyncResponseEnvelopeSchema.Current})."); |
| | 6 | 411 | | AsyncResponseDiagnostics.SetError(activity, "schema_mismatch", schemaError.Message); |
| | 6 | 412 | | if (!tcs.TrySetException(schemaError)) |
| | 2 | 413 | | _logger.LogWarning(schemaError, "TaskCompletionSource already completed for correlationId {Corre |
| | | 414 | | } |
| | 526 | 415 | | else if (!envelope.Success) |
| | | 416 | | { |
| | 9 | 417 | | finished = true; |
| | 9 | 418 | | var remoteFailure = new Exception(envelope.ExceptionMessage ?? "Unknown error during asynchronous pr |
| | 9 | 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. |
| | 2 | 422 | | remoteFailure.Data["RemoteStackTrace"] = RemoteStackTrace.Cap(envelope.ExceptionStackTrace, _opt |
| | | 423 | | |
| | 9 | 424 | | _logger.LogWarning("Received error response for correlationId {CorrelationId}: {ErrorMessage}", corr |
| | 9 | 425 | | AsyncResponseDiagnostics.SetError(activity, "remote_failure", remoteFailure.Message); |
| | 9 | 426 | | if (!tcs.TrySetException(remoteFailure)) |
| | 2 | 427 | | _logger.LogWarning(remoteFailure, "TaskCompletionSource already completed for correlationId {Cor |
| | | 428 | | } |
| | | 429 | | else |
| | | 430 | | { |
| | 517 | 431 | | if (_logger.IsEnabled(LogLevel.Debug)) |
| | 42 | 432 | | _logger.LogDebug("Received response for correlationId {CorrelationId}.", correlationId); |
| | 517 | 433 | | finished = await completionPredicate(envelope.Payload!).ConfigureAwait(false); |
| | 515 | 434 | | if (finished && !tcs.TrySetResult(envelope.Payload!)) |
| | 7 | 435 | | _logger.LogWarning("TaskCompletionSource already completed for correlationId {CorrelationId}.", |
| | | 436 | | } |
| | 536 | 437 | | } |
| | 10 | 438 | | catch (Exception ex) |
| | | 439 | | { |
| | 10 | 440 | | _logger.LogError(ex, "Error processing message on subject {Subject} for correlationId {CorrelationId}.", |
| | 10 | 441 | | finished = true; |
| | 10 | 442 | | AsyncResponseDiagnostics.SetError(activity, ex); |
| | 10 | 443 | | if (!tcs.TrySetException(ex)) |
| | 2 | 444 | | _logger.LogWarning(ex, "TaskCompletionSource already completed for correlationId {CorrelationId}.", |
| | 10 | 445 | | } |
| | | 446 | | finally |
| | | 447 | | { |
| | 548 | 448 | | if (finished) |
| | 427 | 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 | | { |
| | 548 | 461 | | using var correlationScope = AsyncResponseContext.PushCorrelationId(storedCorrelationId); |
| | 548 | 462 | | await ProcessResponseAsync(payload).ConfigureAwait(false); |
| | 548 | 463 | | } |
| | | 464 | | |
| | 548 | 465 | | if (capturedContext is null) |
| | 2 | 466 | | return Process(); |
| | | 467 | | |
| | 546 | 468 | | Task? task = null; |
| | 1092 | 469 | | ExecutionContext.Run(capturedContext, _ => task = Process(), null); |
| | 546 | 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 | | { |
| | 2008 | 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 | | { |
| | 553 | 485 | | await message.ReplyAsync().ConfigureAwait(false); |
| | 551 | 486 | | } |
| | 2 | 487 | | catch (Exception replyEx) |
| | | 488 | | { |
| | 2 | 489 | | _logger.LogDebug(replyEx, "Failed to acknowledge response on subject {Subject}.", subject); |
| | 2 | 490 | | } |
| | | 491 | | |
| | 553 | 492 | | if (message.IsProbe) |
| | | 493 | | continue; |
| | | 494 | | |
| | 548 | 495 | | await ProcessUnderCapturedContextAsync(message.Payload).ConfigureAwait(false); |
| | 548 | 496 | | } |
| | 441 | 497 | | } |
| | 8 | 498 | | catch (Exception ex) |
| | | 499 | | { |
| | 8 | 500 | | _logger.LogError(ex, "Response subscription loop failed for subject {Subject}.", subject); |
| | 8 | 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. |
| | 8 | 510 | | if (!tcs.TrySetException(new NatsConsumeLoopException(ex))) |
| | 4 | 511 | | _logger.LogWarning(ex, "TaskCompletionSource already completed for correlationId {CorrelationId}.", |
| | 8 | 512 | | await CleanupOnceAsync().ConfigureAwait(false); |
| | | 513 | | } |
| | | 514 | | } |
| | | 515 | | |
| | 451 | 516 | | timeoutRegistration = cancellationTokenSource.Token.Register(() => |
| | 451 | 517 | | { |
| | 10 | 518 | | _ = Task.Run(async () => |
| | 10 | 519 | | { |
| | 10 | 520 | | try |
| | 10 | 521 | | { |
| | 10 | 522 | | _logger.LogWarning("Timed out waiting for response for correlationId {CorrelationId}.", correlationI |
| | 8 | 523 | | AsyncResponseDiagnostics.SetError(activity, "timeout", $"Timed out waiting for response for correlat |
| | 8 | 524 | | AsyncResponseDiagnostics.RecordWaiterTimeout("nats"); |
| | 8 | 525 | | await DrainThenCleanupAsync( |
| | 8 | 526 | | new TimeoutException($"Timed out waiting for response for correlationId {correlationId}.")) |
| | 8 | 527 | | .ConfigureAwait(false); |
| | 8 | 528 | | } |
| | 2 | 529 | | catch (Exception ex) |
| | 10 | 530 | | { |
| | 10 | 531 | | // Fire-and-forget: nothing awaits this task, so an escaped fault would vanish. |
| | 2 | 532 | | _logger.LogError(ex, "Error handling waiter timeout for correlationId {CorrelationId}.", correlation |
| | 2 | 533 | | } |
| | 20 | 534 | | }); |
| | 461 | 535 | | }); |
| | | 536 | | |
| | | 537 | | try |
| | | 538 | | { |
| | 451 | 539 | | subscription = await _client.SubscribeAsync(subject, cancellationTokenSource.Token).ConfigureAwait(false); |
| | 902 | 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. |
| | 451 | 552 | | if (Volatile.Read(ref cleanupStarted) == 0) |
| | 451 | 553 | | await _client.FlushAsync(cancellationTokenSource.Token).ConfigureAwait(false); |
| | | 554 | | |
| | 451 | 555 | | var recoveryState = new RecoveryState |
| | 451 | 556 | | { |
| | 451 | 557 | | RegistrationId = registrationId, |
| | 451 | 558 | | ResumeCallback = resumeCallback, |
| | 451 | 559 | | FailureCallback = failureCallback, |
| | 451 | 560 | | CorrelationId = correlationId, |
| | 451 | 561 | | PayloadTypeFullName = typeof(T).FullName, |
| | 451 | 562 | | // The engine's clock, not the ambient one. The watchdog judges staleness as |
| | 451 | 563 | | // "utcNow - RegisteredAtUtc" from whichever host scans, so an unsubstitutable |
| | 451 | 564 | | // app-clock stamp made a skewed host's registrations either never age (skew ahead: |
| | 451 | 565 | | // a genuinely stuck flow stays invisible and the health check stays green) or age |
| | 451 | 566 | | // instantly (skew behind: healthy waits page the operator every scan). The DB |
| | 451 | 567 | | // channels stamp the SERVER clock for exactly this reason; this at least puts the |
| | 451 | 568 | | // stamp and the watchdog's "now" on one substitutable clock, and matches the |
| | 451 | 569 | | // ExpiresAtUtc the recovery store writes for the same registration. |
| | 451 | 570 | | RegisteredAtUtc = _timeProvider.GetUtcNow().UtcDateTime, |
| | 451 | 571 | | Context = _propagation.Capture() |
| | 451 | 572 | | }; |
| | 451 | 573 | | await _recoveryStateStore.SaveAsync(correlationId, recoveryState, _options.RecoveryStateExpiry).ConfigureAwa |
| | 445 | 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 | | { |
| | 6 | 584 | | await _recoveryStateStore.TryDeleteAsync(correlationId, registrationId).ConfigureAwait(false); |
| | 6 | 585 | | } |
| | 0 | 586 | | catch (Exception ex) |
| | | 587 | | { |
| | 0 | 588 | | _logger.LogError(ex, "Post-save recovery-state compensation delete failed for correlationId {Correla |
| | 0 | 589 | | } |
| | | 590 | | } |
| | | 591 | | |
| | 445 | 592 | | _logger.LogDebug("Subscribed to subject {Subject} for correlationId {CorrelationId}.", subject, correlationI |
| | 445 | 593 | | } |
| | 6 | 594 | | catch (Exception ex) when (tcs.Task.IsCompletedSuccessfully |
| | 6 | 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. |
| | 2 | 610 | | _logger.LogWarning(ex, |
| | 2 | 611 | | "Registration step failed after a delivery settled correlationId {CorrelationId}; returning the complete |
| | 2 | 612 | | correlationId); |
| | 2 | 613 | | } |
| | 4 | 614 | | catch (Exception ex) |
| | | 615 | | { |
| | 4 | 616 | | _logger.LogError(ex, "Failed to subscribe to subject {Subject} for correlationId {CorrelationId}.", subject, |
| | 4 | 617 | | AsyncResponseDiagnostics.SetError(activity, "subscribe_failure", ex.Message); |
| | 4 | 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. |
| | 4 | 625 | | throw; |
| | | 626 | | } |
| | | 627 | | |
| | 447 | 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. |
| | 2 | 638 | | throw new InvalidOperationException( |
| | 2 | 639 | | $"The NATS response subscription for correlationId {correlationId} failed before registration completed. |
| | 2 | 640 | | loopFault.InnerException); |
| | | 641 | | } |
| | | 642 | | |
| | | 643 | | try |
| | | 644 | | { |
| | 445 | 645 | | if (Volatile.Read(ref cleanupStarted) == 0) |
| | 439 | 646 | | cancellationTokenSource.CancelAfter(timeout.Value); |
| | 445 | 647 | | } |
| | 0 | 648 | | catch (ObjectDisposedException) |
| | | 649 | | { |
| | | 650 | | // A response completed and cleaned up between the check and CancelAfter. |
| | 0 | 651 | | } |
| | | 652 | | |
| | 890 | 653 | | return new NatsAsyncResponseWaiter<T>(UnwrapConsumeLoopFaults(tcs.Task), () => DrainThenCleanupAsync()); |
| | 2350 | 654 | | } |
| | | 655 | | |
| | | 656 | | // --------------------------------------------------------------------------------------- |
| | | 657 | | // IAsyncResponsePublisher |
| | | 658 | | |
| | | 659 | | /// <inheritdoc/> |
| | | 660 | | public Task SetResponse<T>(T response, string correlationId, CancellationToken cancellationToken = default) where T |
| | 502 | 661 | | => SetResponseCore(response, correlationId, cancellationToken); |
| | | 662 | | |
| | | 663 | | Task IRawAsyncResponsePublisher.SetRawResponse(object? response, string correlationId, CancellationToken cancellatio |
| | 2 | 664 | | => SetResponseCore(response, correlationId, cancellationToken); |
| | | 665 | | |
| | | 666 | | Task IRawAsyncResponsePublisher.SetRawResponseJson(string responseJson, string correlationId, CancellationToken canc |
| | 38 | 667 | | => SetRawResponseJsonCore(responseJson, correlationId, cancellationToken); |
| | | 668 | | |
| | | 669 | | private async Task SetResponseCore<T>(T response, string correlationId, CancellationToken cancellationToken) |
| | | 670 | | { |
| | 504 | 671 | | using var activity = AsyncResponseDiagnostics.StartActivity("asyncresponse.set_response", ActivityKind.Producer) |
| | 504 | 672 | | activity?.SetTag("asyncresponse.channel", "nats"); |
| | 504 | 673 | | AsyncResponseDiagnostics.SetPayloadType(activity, typeof(T)); |
| | | 674 | | |
| | 504 | 675 | | AsyncResponseDiagnostics.SetCorrelationId(activity, correlationId); |
| | | 676 | | |
| | 504 | 677 | | if (CorrelationIdGuard.IsUnpublishable(correlationId, _logger, activity, "the response")) |
| | 3 | 678 | | return; |
| | | 679 | | |
| | 498 | 680 | | var subject = _subjects.ResponseSubject(correlationId); |
| | | 681 | | try |
| | | 682 | | { |
| | 498 | 683 | | var envelope = new AsyncResponseEnvelope<T> { Success = true, Payload = response }; |
| | 498 | 684 | | var json = AsyncResponseEnvelopeJson.Serialize(envelope); |
| | 498 | 685 | | var outcome = await _client.RequestAsync(subject, json, probe: false, _options.DeliveryConfirmationTimeout, |
| | 496 | 686 | | activity?.SetTag("asyncresponse.delivery", outcome.ToString()); |
| | | 687 | | |
| | 496 | 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. |
| | 13 | 692 | | var dispatchResult = await _lostSubscriberDispatcher |
| | 13 | 693 | | .DispatchLostResponses( |
| | 13 | 694 | | _recoveryStateStore, |
| | 13 | 695 | | correlationId, |
| | 13 | 696 | | response, |
| | 13 | 697 | | subject, |
| | 13 | 698 | | cancellationToken, |
| | 13 | 699 | | hasLiveSubscriber: () => HasLiveSubscriberAsync(correlationId, cancellationToken)) |
| | 13 | 700 | | .ConfigureAwait(false); |
| | 13 | 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. |
| | 6 | 706 | | outcome = await _client.RequestAsync(subject, json, probe: false, _options.DeliveryConfirmationTimeo |
| | 6 | 707 | | activity?.SetTag("asyncresponse.delivery", outcome.ToString()); |
| | 6 | 708 | | if (outcome != NatsDeliveryOutcome.NoResponders) |
| | 2 | 709 | | return; |
| | | 710 | | |
| | 4 | 711 | | dispatchResult = await _lostSubscriberDispatcher |
| | 4 | 712 | | .DispatchLostResponses( |
| | 4 | 713 | | _recoveryStateStore, |
| | 4 | 714 | | correlationId, |
| | 4 | 715 | | response, |
| | 4 | 716 | | subject, |
| | 4 | 717 | | cancellationToken, |
| | 4 | 718 | | hasLiveSubscriber: () => HasLiveSubscriberAsync(correlationId, cancellationToken)) |
| | 4 | 719 | | .ConfigureAwait(false); |
| | 4 | 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. |
| | 2 | 732 | | _logger.LogWarning( |
| | 2 | 733 | | "Delivery for correlationId {CorrelationId} found no subscribers twice while the liveness pr |
| | 2 | 734 | | correlationId); |
| | 2 | 735 | | activity?.SetTag("asyncresponse.recovery.liveness_contradiction", true); |
| | 2 | 736 | | throw new InvalidOperationException( |
| | 2 | 737 | | $"NATS delivery for correlationId '{correlationId}' found no responders twice while the live |
| | 2 | 738 | | "reporting a live subscriber; the payload was not delivered and recovery registrations were |
| | 2 | 739 | | "the publish once the waiter's subscription is visible to the publishing endpoint."); |
| | | 740 | | } |
| | | 741 | | } |
| | | 742 | | |
| | 9 | 743 | | AsyncResponseDiagnostics.SetLostSubscriberRoute(activity, dispatchResult.Action, dispatchResult.RouteMix |
| | 9 | 744 | | AsyncResponseDiagnostics.RecordLostSubscriber("response", dispatchResult.Action, dispatchResult.Callback |
| | 9 | 745 | | activity?.SetTag("asyncresponse.recovery.callback_invoked", dispatchResult.CallbackInvoked); |
| | | 746 | | } |
| | 483 | 747 | | else if (_logger.IsEnabled(LogLevel.Debug)) |
| | | 748 | | { |
| | 16 | 749 | | _logger.LogDebug("Published response for correlationId {CorrelationId} on subject {Subject}. PayloadType |
| | | 750 | | } |
| | 492 | 751 | | } |
| | 4 | 752 | | catch (Exception ex) |
| | | 753 | | { |
| | 4 | 754 | | _logger.LogError(ex, "Failed to publish response for correlationId {CorrelationId} on subject {Subject}.", c |
| | 4 | 755 | | AsyncResponseDiagnostics.SetError(activity, ex); |
| | 4 | 756 | | throw; |
| | | 757 | | } |
| | 497 | 758 | | } |
| | | 759 | | |
| | | 760 | | private async Task SetRawResponseJsonCore(string responseJson, string correlationId, CancellationToken cancellationT |
| | | 761 | | { |
| | 38 | 762 | | using var activity = AsyncResponseDiagnostics.StartActivity("asyncresponse.ingress.raw_response", ActivityKind.P |
| | 38 | 763 | | activity?.SetTag("asyncresponse.channel", "nats"); |
| | | 764 | | |
| | 38 | 765 | | AsyncResponseDiagnostics.SetCorrelationId(activity, correlationId); |
| | | 766 | | |
| | 38 | 767 | | if (CorrelationIdGuard.IsUnpublishable(correlationId, _logger, activity, "the raw response", dropContractViolati |
| | 4 | 768 | | return; |
| | | 769 | | |
| | 34 | 770 | | var subject = _subjects.ResponseSubject(correlationId); |
| | | 771 | | try |
| | | 772 | | { |
| | 34 | 773 | | var json = SerializeRawSuccessEnvelope(responseJson); |
| | 34 | 774 | | var outcome = await _client.RequestAsync(subject, json, probe: false, _options.DeliveryConfirmationTimeout, |
| | 32 | 775 | | activity?.SetTag("asyncresponse.delivery", outcome.ToString()); |
| | | 776 | | |
| | 32 | 777 | | if (outcome == NatsDeliveryOutcome.NoResponders) |
| | | 778 | | { |
| | 23 | 779 | | var response = new RawJsonResponse(responseJson).DeserializeUntyped(); |
| | | 780 | | |
| | 23 | 781 | | var dispatchResult = await _lostSubscriberDispatcher |
| | 23 | 782 | | .DispatchLostResponses( |
| | 23 | 783 | | _recoveryStateStore, |
| | 23 | 784 | | correlationId, |
| | 23 | 785 | | response, |
| | 23 | 786 | | subject, |
| | 23 | 787 | | cancellationToken, |
| | 23 | 788 | | hasLiveSubscriber: () => HasLiveSubscriberAsync(correlationId, cancellationToken)) |
| | 23 | 789 | | .ConfigureAwait(false); |
| | 21 | 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. |
| | 6 | 795 | | outcome = await _client.RequestAsync(subject, json, probe: false, _options.DeliveryConfirmationTimeo |
| | 6 | 796 | | activity?.SetTag("asyncresponse.delivery", outcome.ToString()); |
| | 6 | 797 | | if (outcome != NatsDeliveryOutcome.NoResponders) |
| | 2 | 798 | | return; |
| | | 799 | | |
| | 4 | 800 | | dispatchResult = await _lostSubscriberDispatcher |
| | 4 | 801 | | .DispatchLostResponses( |
| | 4 | 802 | | _recoveryStateStore, |
| | 4 | 803 | | correlationId, |
| | 4 | 804 | | response, |
| | 4 | 805 | | subject, |
| | 4 | 806 | | cancellationToken, |
| | 4 | 807 | | hasLiveSubscriber: () => HasLiveSubscriberAsync(correlationId, cancellationToken)) |
| | 4 | 808 | | .ConfigureAwait(false); |
| | 4 | 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. |
| | 2 | 821 | | _logger.LogWarning( |
| | 2 | 822 | | "Delivery for correlationId {CorrelationId} found no subscribers twice while the liveness pr |
| | 2 | 823 | | correlationId); |
| | 2 | 824 | | activity?.SetTag("asyncresponse.recovery.liveness_contradiction", true); |
| | 2 | 825 | | throw new InvalidOperationException( |
| | 2 | 826 | | $"NATS delivery for correlationId '{correlationId}' found no responders twice while the live |
| | 2 | 827 | | "reporting a live subscriber; the payload was not delivered and recovery registrations were |
| | 2 | 828 | | "the publish once the waiter's subscription is visible to the publishing endpoint."); |
| | | 829 | | } |
| | | 830 | | } |
| | | 831 | | |
| | 17 | 832 | | AsyncResponseDiagnostics.SetLostSubscriberRoute(activity, dispatchResult.Action, dispatchResult.RouteMix |
| | 17 | 833 | | AsyncResponseDiagnostics.RecordLostSubscriber("response", dispatchResult.Action, dispatchResult.Callback |
| | 17 | 834 | | activity?.SetTag("asyncresponse.recovery.callback_invoked", dispatchResult.CallbackInvoked); |
| | 17 | 835 | | } |
| | 9 | 836 | | else if (_logger.IsEnabled(LogLevel.Debug)) |
| | | 837 | | { |
| | 4 | 838 | | _logger.LogDebug("Published raw response for correlationId {CorrelationId} on subject {Subject}. Outcome |
| | | 839 | | } |
| | 26 | 840 | | } |
| | 6 | 841 | | catch (Exception ex) |
| | | 842 | | { |
| | 6 | 843 | | _logger.LogError(ex, "Failed to publish raw response for correlationId {CorrelationId} on subject {Subject}. |
| | 6 | 844 | | AsyncResponseDiagnostics.SetError(activity, ex); |
| | 6 | 845 | | throw; |
| | | 846 | | } |
| | 32 | 847 | | } |
| | | 848 | | |
| | | 849 | | /// <inheritdoc/> |
| | | 850 | | public async Task SetException(Exception exception, string correlationId, CancellationToken cancellationToken = defa |
| | | 851 | | { |
| | 24 | 852 | | ArgumentNullException.ThrowIfNull(exception); |
| | | 853 | | |
| | 24 | 854 | | using var activity = AsyncResponseDiagnostics.StartActivity("asyncresponse.set_exception", ActivityKind.Producer |
| | 24 | 855 | | activity?.SetTag("asyncresponse.channel", "nats"); |
| | 24 | 856 | | activity?.SetTag("asyncresponse.exception_type", exception.GetType().FullName ?? exception.GetType().Name); |
| | | 857 | | |
| | 24 | 858 | | AsyncResponseDiagnostics.SetCorrelationId(activity, correlationId); |
| | | 859 | | |
| | 24 | 860 | | if (CorrelationIdGuard.IsUnpublishable(correlationId, _logger, activity, "the exception", exception)) |
| | 3 | 861 | | return; |
| | | 862 | | |
| | 20 | 863 | | var subject = _subjects.ResponseSubject(correlationId); |
| | | 864 | | try |
| | | 865 | | { |
| | 20 | 866 | | var envelope = new AsyncResponseEnvelope<object> |
| | 20 | 867 | | { |
| | 20 | 868 | | Success = false, |
| | 20 | 869 | | ExceptionMessage = exception.Message, |
| | 20 | 870 | | ExceptionStackTrace = RemoteStackTrace.ForWire(exception.StackTrace, _options.IncludeRemoteStackTrace, _ |
| | 20 | 871 | | Payload = null |
| | 20 | 872 | | }; |
| | 20 | 873 | | var json = AsyncResponseEnvelopeJson.Serialize(envelope); |
| | 20 | 874 | | var outcome = await _client.RequestAsync(subject, json, probe: false, _options.DeliveryConfirmationTimeout, |
| | 18 | 875 | | activity?.SetTag("asyncresponse.delivery", outcome.ToString()); |
| | | 876 | | |
| | 18 | 877 | | if (outcome == NatsDeliveryOutcome.NoResponders) |
| | | 878 | | { |
| | | 879 | | // Nobody was listening: exception envelopes always go to the failure callback. |
| | 11 | 880 | | var dispatchResult = await _lostSubscriberDispatcher |
| | 11 | 881 | | .DispatchLostExceptions( |
| | 11 | 882 | | _recoveryStateStore, |
| | 11 | 883 | | correlationId, |
| | 11 | 884 | | exception, |
| | 11 | 885 | | subject, |
| | 11 | 886 | | cancellationToken, |
| | 11 | 887 | | hasLiveSubscriber: () => HasLiveSubscriberAsync(correlationId, cancellationToken)) |
| | 11 | 888 | | .ConfigureAwait(false); |
| | 11 | 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. |
| | 6 | 894 | | outcome = await _client.RequestAsync(subject, json, probe: false, _options.DeliveryConfirmationTimeo |
| | 6 | 895 | | activity?.SetTag("asyncresponse.delivery", outcome.ToString()); |
| | 6 | 896 | | if (outcome != NatsDeliveryOutcome.NoResponders) |
| | 2 | 897 | | return; |
| | | 898 | | |
| | 4 | 899 | | dispatchResult = await _lostSubscriberDispatcher |
| | 4 | 900 | | .DispatchLostExceptions( |
| | 4 | 901 | | _recoveryStateStore, |
| | 4 | 902 | | correlationId, |
| | 4 | 903 | | exception, |
| | 4 | 904 | | subject, |
| | 4 | 905 | | cancellationToken, |
| | 4 | 906 | | hasLiveSubscriber: () => HasLiveSubscriberAsync(correlationId, cancellationToken)) |
| | 4 | 907 | | .ConfigureAwait(false); |
| | 4 | 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. |
| | 2 | 920 | | _logger.LogWarning( |
| | 2 | 921 | | "Delivery for correlationId {CorrelationId} found no subscribers twice while the liveness pr |
| | 2 | 922 | | correlationId); |
| | 2 | 923 | | activity?.SetTag("asyncresponse.recovery.liveness_contradiction", true); |
| | 2 | 924 | | throw new InvalidOperationException( |
| | 2 | 925 | | $"NATS delivery for correlationId '{correlationId}' found no responders twice while the live |
| | 2 | 926 | | "reporting a live subscriber; the payload was not delivered and recovery registrations were |
| | 2 | 927 | | "the publish once the waiter's subscription is visible to the publishing endpoint."); |
| | | 928 | | } |
| | | 929 | | } |
| | | 930 | | |
| | 7 | 931 | | activity?.SetTag("asyncresponse.recovery.callback_invoked", dispatchResult.CallbackInvoked); |
| | 7 | 932 | | AsyncResponseDiagnostics.RecordLostSubscriber("exception", action: RecoveryAction.Fail, dispatchResult.C |
| | | 933 | | } |
| | 7 | 934 | | else if (_logger.IsEnabled(LogLevel.Debug)) |
| | | 935 | | { |
| | 6 | 936 | | _logger.LogDebug("Published exception response for correlationId {CorrelationId} on subject {Subject}. O |
| | | 937 | | } |
| | 14 | 938 | | } |
| | 4 | 939 | | catch (Exception ex) |
| | | 940 | | { |
| | 4 | 941 | | _logger.LogError(ex, "Failed to publish exception response for correlationId {CorrelationId} on subject {Sub |
| | 4 | 942 | | AsyncResponseDiagnostics.SetError(activity, ex); |
| | 4 | 943 | | throw; |
| | | 944 | | } |
| | 19 | 945 | | } |
| | | 946 | | |
| | | 947 | | // --------------------------------------------------------------------------------------- |
| | | 948 | | // IActiveSubscriberProbe |
| | | 949 | | |
| | | 950 | | /// <inheritdoc/> |
| | | 951 | | public async ValueTask<long> CountActiveSubscribersAsync(string correlationId, CancellationToken cancellationToken = |
| | | 952 | | { |
| | 78 | 953 | | if (string.IsNullOrWhiteSpace(correlationId)) |
| | 2 | 954 | | return 0L; |
| | | 955 | | |
| | 76 | 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. |
| | 76 | 967 | | var outcome = await _client.RequestAsync(subject, payload: null, probe: true, _options.PresenceProbeTimeout, |
| | 70 | 968 | | return outcome switch |
| | 70 | 969 | | { |
| | 29 | 970 | | NatsDeliveryOutcome.Replied => 1L, |
| | 39 | 971 | | NatsDeliveryOutcome.NoResponders => 0L, |
| | 2 | 972 | | _ => -1L |
| | 70 | 973 | | }; |
| | | 974 | | } |
| | 2 | 975 | | catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) |
| | | 976 | | { |
| | 2 | 977 | | throw; |
| | | 978 | | } |
| | 4 | 979 | | catch (Exception ex) |
| | | 980 | | { |
| | 4 | 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. |
| | 4 | 985 | | return -1L; |
| | | 986 | | } |
| | 76 | 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 | | { |
| | 59 | 998 | | var subscribers = await CountActiveSubscribersAsync(correlationId, cancellationToken).ConfigureAwait(false); |
| | 59 | 999 | | if (subscribers < 0) |
| | | 1000 | | { |
| | 2 | 1001 | | throw new InvalidOperationException( |
| | 2 | 1002 | | $"NATS subscriber liveness for correlationId '{correlationId}' could not be probed."); |
| | | 1003 | | } |
| | | 1004 | | |
| | 57 | 1005 | | return subscribers > 0; |
| | 57 | 1006 | | } |
| | | 1007 | | |
| | | 1008 | | private static string SerializeRawSuccessEnvelope(string payloadJson) |
| | | 1009 | | { |
| | 34 | 1010 | | JsonSafety.ThrowIfClearlyNotJson(payloadJson); |
| | | 1011 | | |
| | 34 | 1012 | | var buffer = new ArrayBufferWriter<byte>(); |
| | 34 | 1013 | | using (var writer = new Utf8JsonWriter(buffer)) |
| | | 1014 | | { |
| | 34 | 1015 | | writer.WriteStartObject(); |
| | 34 | 1016 | | writer.WriteNumber("SchemaVersion", AsyncResponseEnvelopeSchema.Current); |
| | 34 | 1017 | | writer.WriteBoolean("Success", true); |
| | 34 | 1018 | | writer.WritePropertyName("Payload"); |
| | 34 | 1019 | | writer.WriteRawValue(payloadJson); |
| | 34 | 1020 | | writer.WriteNull("ExceptionMessage"); |
| | 34 | 1021 | | writer.WriteNull("ExceptionStackTrace"); |
| | 34 | 1022 | | writer.WriteEndObject(); |
| | 34 | 1023 | | } |
| | | 1024 | | |
| | 34 | 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> |
| | | 1034 | | internal sealed class NatsConsumeLoopException(Exception inner) |
| | | 1035 | | : Exception("The NATS response subscription loop failed.", inner); |