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

Information
Class: AsyncResponse.Channels.Redis.RedisAsyncResponseChannel
Assembly: AsyncResponse.Channels.Redis
File(s): /_/src/Channels/AsyncResponse.Channels.Redis/RedisAsyncResponseChannel.cs
Line coverage
95%
Covered lines: 532
Uncovered lines: 23
Coverable lines: 555
Total lines: 1177
Line coverage: 95.8%
Branch coverage
91%
Covered branches: 159
Total branches: 174
Branch coverage: 91.3%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%44100%
TrackRetirement(...)100%11100%
DisposeAsync()100%22100%
CreateResponseWaiter(...)100%11100%
CreateRecoverableResponseWaiter(...)100%11100%
CreateResponseWaiterCore()95.83%252489.61%
.ctor(...)100%11100%
get_Id()100%11100%
get_ChannelName()100%11100%
get_ResponseTask()100%11100%
get_CleanupStarted()100%11100%
get_Subscription()100%11100%
get_ExecutorRegistered()100%11100%
RegisterTimeoutCallback()100%22100%
ArmTimeout(...)100%2266.66%
OnTimeout()100%11100%
HandleMessageAsync(...)100%44100%
OnOverloadedAsync()25%4478.57%
ProcessAsync()100%11100%
ProcessUnderCapturedContextAsync(...)100%22100%
ProcessMessageAsync()93.33%303097.87%
CleanupOnceAsync()100%44100%
DrainThenCleanupAsync()90%101095.83%
CleanupCoreAsync()87.5%8880.76%
<CleanupCoreAsync()100%1150%
UnsubscribeQuietlyAsync()100%11100%
SetResponse(...)100%11100%
AsyncResponse.IRawAsyncResponsePublisher.SetRawResponse(...)100%11100%
AsyncResponse.IRawAsyncResponsePublisher.SetRawResponseJson(...)100%11100%
RetireExecutorBoundedAsync()100%11100%
SetResponseCore()90.9%2222100%
SetRawResponseJsonCore()90.9%2222100%
SetException()88.46%2626100%
CountActiveSubscribersAsync()100%1010100%
HasLiveSubscriberAsync()100%22100%
SerializeRawSuccessEnvelope(...)100%11100%

File(s)

/_/src/Channels/AsyncResponse.Channels.Redis/RedisAsyncResponseChannel.cs

#LineLine coverage
 1using Microsoft.Extensions.DependencyInjection;
 2using Microsoft.Extensions.Logging;
 3using Microsoft.Extensions.Options;
 4using StackExchange.Redis;
 5using System.Buffers;
 6using System.Collections.Concurrent;
 7using System.Diagnostics;
 8using System.Text;
 9using System.Text.Json;
 10
 11namespace AsyncResponse.Channels.Redis;
 12
 13/// <summary>
 14/// Redis-backed response channel:
 15/// <list type="bullet">
 16/// <item><description>Publishes responses to Redis pub/sub channels keyed by correlation id.</description></item>
 17/// <item><description>Subscribes waiters to those channels with per-channel serialized handling.</description></item>
 18/// <item><description>Persists <see cref="RecoveryState"/> so responses arriving after the waiter
 19/// died (e.g. a redeploy) are routed through the lost-subscriber dispatcher, which asks the payload's
 20/// OnRecovery and invokes the resume or failure callback with the materialized payload
 21/// (or keeps the registration armed for a checkpoint).</description></item>
 22/// </list>
 23/// </summary>
 24internal sealed class RedisAsyncResponseChannel : IAsyncResponsePublisher, IRawAsyncResponsePublisher, IRecoverableAsync
 25{
 26
 27    private readonly ISubscriber _subscriber;
 28    private readonly IRedisChannelSubscriber _channelSubscriber;
 29    private readonly IConnectionMultiplexer _multiplexer;
 30    private readonly IRecoveryStateStore _recoveryStateStore;
 31    private readonly AsyncResponseContextPropagation _propagation;
 32    private readonly LostSubscriberCallbackDispatcher _lostSubscriberDispatcher;
 33    private readonly RedisKeySchema _keys;
 34    private readonly RedisAsyncResponseOptions _options;
 35    private readonly ILogger<RedisAsyncResponseChannel> _logger;
 36    private readonly TimeProvider _timeProvider;
 37
 38    private readonly SerialExecutorRegistry _executors;
 39
 40    /// <summary>Creates a Redis-backed async-response channel.</summary>
 55241    public RedisAsyncResponseChannel(
 55242        IServiceScopeFactory scopeFactory,
 55243        IConnectionMultiplexer multiplexer,
 55244        IRecoveryStateStore recoveryStateStore,
 55245        IOptions<RedisAsyncResponseOptions> options,
 55246        AsyncResponseContextPropagation propagation,
 55247        ILogger<RedisAsyncResponseChannel> logger,
 55248        IRedisChannelSubscriber? channelSubscriber = null,
 55249        TimeProvider? timeProvider = null)
 50    {
 55251        _timeProvider = timeProvider ?? TimeProvider.System;
 55252        _subscriber = multiplexer.GetSubscriber();
 55253        _channelSubscriber = channelSubscriber ?? new RedisChannelMessageQueueSubscriber(_subscriber);
 55254        _multiplexer = multiplexer;
 55255        _recoveryStateStore = recoveryStateStore;
 55256        _propagation = propagation;
 55257        _options = options.Value;
 55258        _options.Validate();
 55259        _keys = new RedisKeySchema(_options.KeyPrefix);
 55260        _logger = logger;
 55261        _lostSubscriberDispatcher = new LostSubscriberCallbackDispatcher(scopeFactory, propagation, logger, _timeProvide
 55262        _executors = new SerialExecutorRegistry(logger, timeProvider: _timeProvider);
 55263    }
 64
 65    // Executor retirements scheduled off the cleanup path (see CleanupCoreAsync). TRACKED, as
 66    // DbChannelShared does: untracked, a retirement could still be inside its drain budget — a
 67    // user completion predicate mid-flight — when the host tore down the logger and Main
 68    // returned, and the pool thread running it was killed with the predicate's side effects
 69    // half-applied. Keyed by the task itself and self-evicting.
 55270    private readonly ConcurrentDictionary<Task, byte> _pendingRetirements = new();
 71
 72    private void TrackRetirement(Task retirement)
 73    {
 42974        _pendingRetirements[retirement] = 0;
 42975        _ = retirement.ContinueWith(
 42976            static (completed, state) => ((ConcurrentDictionary<Task, byte>)state!).TryRemove(completed, out _),
 42977            _pendingRetirements,
 42978            CancellationToken.None,
 42979            TaskContinuationOptions.ExecuteSynchronously,
 42980            TaskScheduler.Default);
 42981    }
 82
 83    /// <summary>
 84    /// Joins every executor retirement still in flight, so container disposal at host shutdown
 85    /// means "every executor is retired" rather than "every retirement was started". The bodies
 86    /// swallow, so this cannot throw; the drain budgets inside RemoveAsync bound how long it takes.
 87    /// </summary>
 88    public async ValueTask DisposeAsync()
 89    {
 206290        var retirements = _pendingRetirements.Keys.ToArray();
 206291        if (retirements.Length > 0)
 2592            await Task.WhenAll(retirements).ConfigureAwait(false);
 206293    }
 94
 95    // ---------------------------------------------------------------------------------------
 96    // IAsyncResponseSubscriber / IRecoverableAsyncResponseSubscriber
 97
 98    /// <inheritdoc/>
 99    public Task<IAsyncResponseWaiter<T>> CreateResponseWaiter<T>(
 100        string correlationId,
 101        Func<T, ValueTask<bool>>? completionPredicate = null,
 102        TimeSpan? timeout = null) where T : IAsyncResponsePayload
 215103        => CreateResponseWaiterCore(
 215104            correlationId,
 215105            resumeCallback: null,
 215106            failureCallback: null,
 215107            completionPredicate,
 215108            timeout);
 109
 110    /// <inheritdoc/>
 111    public Task<IAsyncResponseWaiter<T>> CreateRecoverableResponseWaiter<T>(
 112        string correlationId,
 113        ReflectionCallDto? resumeCallback = null,
 114        ReflectionCallDto? failureCallback = null,
 115        Func<T, ValueTask<bool>>? completionPredicate = null,
 116        TimeSpan? timeout = null) where T : IAsyncResponsePayload
 226117        => CreateResponseWaiterCore(
 226118            correlationId,
 226119            resumeCallback,
 226120            failureCallback,
 226121            completionPredicate,
 226122            timeout);
 123
 124    private async Task<IAsyncResponseWaiter<T>> CreateResponseWaiterCore<T>(
 125        string correlationId,
 126        ReflectionCallDto? resumeCallback,
 127        ReflectionCallDto? failureCallback,
 128        Func<T, ValueTask<bool>>? completionPredicate,
 129        TimeSpan? timeout) where T : IAsyncResponsePayload
 130    {
 441131        CorrelationIdGuard.ThrowIfUnusable(correlationId);
 132
 133        // Recovery callbacks only make sense if the payload can say whether a late response should
 134        // resume or fail the flow. On this durable channel that decision is real (it survives a
 135        // redeploy), so require the override rather than letting the conservative default silently
 136        // route every recovered response to the failure callback. The in-memory channel, which
 137        // cannot recover across a process restart, is deliberately not subject to this check.
 435138        if ((resumeCallback is not null || failureCallback is not null)
 435139            && !AsyncResponsePayloadReflection.OverridesOnRecovery(typeof(T)))
 140        {
 4141            throw new InvalidOperationException(
 4142                $"Payload type '{typeof(T)}' registers lost-subscriber recovery callbacks on the Redis channel " +
 4143                $"but does not override {nameof(IAsyncResponsePayload)}.{nameof(IAsyncResponsePayload.OnRecovery)}(). " 
 4144                "Override it to declare what each response does to the flow — RecoveryAction.Resume, " +
 4145                "RecoveryAction.Fail, or RecoveryAction.KeepWaiting for non-terminal checkpoints; the durable " +
 4146                "channel needs this to route a response that arrives after the waiter was lost.");
 147        }
 148
 149        // default: first envelope completes the wait
 583150        completionPredicate ??= _ => new ValueTask<bool>(true);
 151
 152        // Default timeout aligned with the recovery-state expiry: an infinite wait is never
 153        // meaningful, because once the recovery state expires the correlation id has no recovery
 154        // anyway. Timing out routes the flow through its normal failure handling instead of
 155        // leaving it stuck forever.
 431156        timeout ??= _options.DefaultTimeout ?? _options.RecoveryStateExpiry;
 157        // BEFORE any side effect: an unsupported resolved timeout (non-positive, or past the
 158        // ~49.7-day BCL timer ceiling) used to throw only at timer arming — after the
 159        // subscription and recovery state existed, leaking both — and zero used to slip through
 160        // on some channels entirely, insta-timing-out a fully registered waiter.
 431161        AsyncResponseChannelOptions.EnsureWaiterTimeoutSupported(timeout.Value);
 162
 163        // Capture the subscribe-time ExecutionContext so app AsyncLocals (trace, principal, logging
 164        // scope) flow into the message handler, which runs on a foreign Redis subscriber thread.
 429165        var capturedContext = ExecutionContext.Capture();
 429166        var channel = _keys.Channel(correlationId);
 167
 429168        var activity = AsyncResponseDiagnostics.StartActivity("asyncresponse.wait", correlationId: correlationId);
 429169        activity?.SetTag("asyncresponse.channel", "redis");
 429170        AsyncResponseDiagnostics.SetPayloadType(activity, typeof(T));
 429171        activity?.SetTag("asyncresponse.timeout_seconds", timeout.Value.TotalSeconds);
 172
 429173        if (_logger.IsEnabled(LogLevel.Debug))
 6174            _logger.LogDebug("Waiting for response on correlationId {CorrelationId} with timeout {Timeout}.", correlatio
 175
 429176        var subscription = new RedisSubscription<T>(
 429177            this,
 429178            correlationId,
 429179            channel,
 429180            registrationId: Guid.NewGuid(),
 429181            completionPredicate,
 429182            capturedContext,
 429183            activity);
 429184        var channelName = subscription.ChannelName;
 185
 186        // Single-use cancellation token implementing the timeout. The timer is armed only
 187        // after subscribe + recovery-state save succeeds, but the callback is registered before
 188        // subscribing so a very fast terminal message can still clean up safely.
 429189        subscription.RegisterTimeoutCallback();
 190
 191        try
 192        {
 193            // Register the executor channel BEFORE the server-side SUBSCRIBE completes: the
 194            // subscriber attaches its message pump inside SubscribeAsync, so deliveries can start
 195            // before it returns, and on a correlation id reused within the tombstone lifetime the
 196            // registry would silently drop them as retirement stragglers until this registration
 197            // is visible. The subscribe-failure path below retires it again.
 429198            _executors.OnSubscriptionRegistered(channelName);
 429199            subscription.ExecutorRegistered = true;
 429200            subscription.Subscription = await _channelSubscriber.SubscribeAsync(channel, subscription.HandleMessageAsync
 427201            if (subscription.CleanupStarted)
 202            {
 203                // A message pumped inside SubscribeAsync completed the waiter and ran cleanup to
 204                // the end before this assignment existed — its unsubscribe saw a null
 205                // subscription and the latched cleanup never re-runs. Compensate here, or the
 206                // server-side subscription outlives the waiter: NUMSUB and publish keep counting
 207                // a live waiter, and lost-subscriber recovery is suppressed for this correlation
 208                // id until process exit. Best-effort: the waiter already holds its response, so a
 209                // teardown fault must not fail the create.
 210                try
 211                {
 0212                    await subscription.UnsubscribeQuietlyAsync(subscription.Subscription).WaitAsync(_options.DisposalDra
 0213                }
 0214                catch (Exception ex)
 215                {
 0216                    _logger.LogError(ex, "Post-registration unsubscribe for channel {Channel} failed; the subscription m
 0217                }
 218            }
 219
 427220            var recoveryState = new RecoveryState
 427221            {
 427222                RegistrationId = subscription.Id,
 427223                ResumeCallback = resumeCallback,
 427224                FailureCallback = failureCallback,
 427225                CorrelationId = correlationId,
 427226                PayloadTypeFullName = typeof(T).FullName,
 427227                // The engine's clock, not the ambient one. The watchdog judges staleness as
 427228                // "utcNow - RegisteredAtUtc" from whichever host scans, so an unsubstitutable
 427229                // app-clock stamp made a skewed host's registrations either never age (skew ahead:
 427230                // a genuinely stuck flow stays invisible and the health check stays green) or age
 427231                // instantly (skew behind: healthy waits page the operator every scan). The DB
 427232                // channels stamp the SERVER clock for exactly this reason; this at least puts the
 427233                // stamp and the watchdog's "now" on one substitutable clock, and matches the
 427234                // ExpiresAtUtc the recovery store writes for the same registration.
 427235                RegisteredAtUtc = _timeProvider.GetUtcNow().UtcDateTime,
 427236                Context = _propagation.Capture()
 427237            };
 427238            await _recoveryStateStore.SaveAsync(correlationId, recoveryState, _options.RecoveryStateExpiry).ConfigureAwa
 425239            if (subscription.CleanupStarted)
 240            {
 241                // A terminal delivery started cleanup while this registration was still being
 242                // written: cleanup's delete ran before the save committed, so the save just
 243                // orphaned a callback-armed registration that would resurrect recovery for a wait
 244                // that already reached a terminal state. Compensate with a second delete
 245                // (mirrors the in-memory channel's post-save check). Best-effort: TTL and the
 246                // watchdog back a failed delete.
 247                try
 248                {
 2249                    await _recoveryStateStore.TryDeleteAsync(correlationId, subscription.Id).ConfigureAwait(false);
 2250                }
 0251                catch (Exception ex)
 252                {
 0253                    _logger.LogError(ex, "Post-save recovery-state compensation delete failed for correlationId {Correla
 0254                }
 255            }
 256
 425257            _logger.LogDebug("Subscribed to channel {Channel} for correlationId {CorrelationId}.", channelName, correlat
 425258        }
 4259        catch (Exception ex) when (subscription.ResponseTask.IsCompletedSuccessfully || subscription.ResponseTask.IsFaul
 260        {
 261            // The wait already settled: a delivery completed the waiter while this registration
 262            // step was still in flight (cleanup marks cleanupStarted just after setting the task,
 263            // so the task is the race-free signal), and the step — the recovery-state save — then
 264            // failed. The response in hand outranks the builder's "throw so the trigger never
 265            // fires" contract: rethrowing would discard a delivered response, the exact loss this
 266            // library exists to prevent, and the success path for this same interleaving already
 267            // returns the completed waiter. Cleanup runs on the delivery path, so nothing is
 268            // leaked; a save that still committed is compensated above or expires via TTL, with
 269            // the recovery watchdog behind it. The filter demands an actual settlement (result
 270            // or fault): a canceled task means NO response was delivered — e.g. a future
 271            // channel-wide teardown canceling in-flight registrations — and takes the rethrow
 272            // path below.
 2273            _logger.LogWarning(ex,
 2274                "Registration step failed after a delivery settled correlationId {CorrelationId}; returning the complete
 2275                correlationId);
 2276        }
 2277        catch (Exception ex)
 278        {
 2279            _logger.LogError(ex, "Failed to subscribe to channel {Channel} for correlationId {CorrelationId}.", channelN
 2280            AsyncResponseDiagnostics.SetError(activity, "subscribe_failure", ex.Message);
 2281            await subscription.DrainThenCleanupAsync().ConfigureAwait(false);
 282
 283            // Rethrow instead of returning a pre-faulted waiter: the builder's contract is that
 284            // the trigger runs only once the subscription AND recovery state exist. A returned
 285            // waiter would still let the trigger fire the remote operation with no registration
 286            // left to receive (or recover) its response. Cleanup leaves nothing behind and cancels
 287            // the response task rather than faulting it, so no unobserved fault lingers.
 2288            throw;
 289        }
 290
 427291        subscription.ArmTimeout(timeout.Value);
 292
 854293        return new RedisAsyncResponseWaiter<T>(subscription.ResponseTask, () => subscription.DrainThenCleanupAsync());
 427294    }
 295
 296    /// <summary>
 297    /// Per-waiter subscription state and lifecycle. A concrete class rather than closures over the
 298    /// creating method: the message handler and timeout callback live for the whole wait — days
 299    /// for a durable-flow await — and must retain only these fields, not a display class holding
 300    /// every local of the registration scope. The channel-name string is rendered once here;
 301    /// dispatch and cleanup key the executor registry with it instead of re-rendering the
 302    /// <see cref="RedisChannel"/> per message.
 303    /// </summary>
 304    private sealed class RedisSubscription<T> where T : IAsyncResponsePayload
 305    {
 306        private readonly RedisAsyncResponseChannel _owner;
 307        private readonly string _correlationId;
 308        private readonly Func<T, ValueTask<bool>> _completionPredicate;
 309        private readonly ExecutionContext? _capturedContext;
 310        private readonly Activity? _activity;
 429311        private readonly TaskCompletionSource<T> _tcs = new(TaskCreationOptions.RunContinuationsAsynchronously);
 312
 313        // Single-use cancellation token implementing the waiter timeout. Clock-injected
 314        // (DbChannelShared parity): CancelAfter on a default CTS is bound to the system clock,
 315        // so a virtual clock could never fire a production-sized waiter timeout on this channel.
 316        private readonly CancellationTokenSource _cancellationTokenSource;
 317        private CancellationTokenRegistration _timeoutRegistration;
 318
 319        // Ensures unsubscribe, recovery-state delete, timeout disposal, and executor cleanup
 320        // happen once no matter whether completion, timeout, or waiter disposal got there first.
 321        private int _cleanupStarted;
 322
 323        // Set by the overload fault: every message still queued behind it is skipped unprocessed.
 324        private int _overloaded;
 429325        private readonly object _cleanupGate = new();
 326        private Task? _cleanupTask;
 327
 429328        public RedisSubscription(
 429329            RedisAsyncResponseChannel owner,
 429330            string correlationId,
 429331            RedisChannel channel,
 429332            Guid registrationId,
 429333            Func<T, ValueTask<bool>> completionPredicate,
 429334            ExecutionContext? capturedContext,
 429335            Activity? activity)
 336        {
 429337            _owner = owner;
 429338            _cancellationTokenSource = new CancellationTokenSource(Timeout.InfiniteTimeSpan, owner._timeProvider);
 429339            _correlationId = correlationId;
 429340            ChannelName = channel.ToString()!;
 429341            Id = registrationId;
 429342            _completionPredicate = completionPredicate;
 429343            _capturedContext = capturedContext;
 429344            _activity = activity;
 429345        }
 346
 347        /// <summary>Per-waiter registration id used for recovery-state cleanup.</summary>
 858348        public Guid Id { get; }
 4871349        public string ChannelName { get; }
 433350        public Task<T> ResponseTask => _tcs.Task;
 852351        public bool CleanupStarted => Volatile.Read(ref _cleanupStarted) != 0;
 1283352        public IRedisChannelSubscription? Subscription { get; set; }
 889353        public bool ExecutorRegistered { get; set; }
 354
 355        /// <summary>
 356        /// Registers the timeout callback on the (not yet armed) token; the creator registers it
 357        /// before subscribing so a very fast terminal message can still clean up safely.
 358        /// </summary>
 359        public void RegisterTimeoutCallback()
 429360            => _timeoutRegistration = _cancellationTokenSource.Token.Register(
 439361                static state => ((RedisSubscription<T>)state!).OnTimeout(), this);
 362
 363        /// <summary>Arms the timeout once registration has succeeded; a no-op after cleanup started.</summary>
 364        public void ArmTimeout(TimeSpan timeout)
 365        {
 366            try
 367            {
 427368                if (Volatile.Read(ref _cleanupStarted) == 0)
 423369                    _cancellationTokenSource.CancelAfter(timeout);
 427370            }
 0371            catch (ObjectDisposedException)
 372            {
 373                // A response completed and cleaned up between the check and CancelAfter.
 0374            }
 427375        }
 376
 377        private void OnTimeout()
 10378            => _ = Task.Run(async () =>
 10379            {
 10380                try
 10381                {
 10382                    _owner._logger.LogWarning("Timed out waiting for response for correlationId {CorrelationId}.", _corr
 8383                    AsyncResponseDiagnostics.SetError(_activity, "timeout", $"Timed out waiting for response for correla
 8384                    AsyncResponseDiagnostics.RecordWaiterTimeout("redis");
 8385                    await DrainThenCleanupAsync(
 8386                        new TimeoutException($"Timed out waiting for response for correlationId {_correlationId}."))
 8387                        .ConfigureAwait(false);
 8388                }
 2389                catch (Exception ex)
 10390                {
 10391                    // Fire-and-forget: nothing awaits this task, so an escaped fault would vanish.
 2392                    _owner._logger.LogError(ex, "Error handling waiter timeout for correlationId {CorrelationId}.", _cor
 2393                }
 20394            });
 395
 396        /// <summary>
 397        /// Receives pub/sub messages from the async subscription and admits them to the
 398        /// per-channel serial executor WITHOUT waiting for capacity. Redis pub/sub is
 399        /// fire-and-forget: the publisher is never backpressured, and the SDK's
 400        /// <c>ChannelMessageQueue</c> behind this callback is unbounded — so an earlier version
 401        /// that awaited executor admission here did not slow anything down, it only moved the
 402        /// backlog from the bounded executor into that unbounded SDK queue, where a progress-message
 403        /// burst behind a slow <c>Until</c> predicate could grow process memory until failure. The
 404        /// executor's capacity (<see cref="ChannelSerialExecutor.DefaultCapacity"/> messages per
 405        /// correlation id) is now the whole buffer: a message that finds it full faults the wait
 406        /// as indeterminate (<see cref="OnOverloadedAsync"/>) instead of being buffered without
 407        /// bound — and never silently dropped, since a terminal response may be among the queued ones.
 408        /// </summary>
 409        public Task HandleMessageAsync(RedisChannel messageChannel, RedisValue messageValue)
 410        {
 411            // The registry coordinates create/enqueue/retire under one lock, so the message is never
 412            // enqueued onto an executor that is concurrently being torn down (no lost messages) and a
 413            // correlation-id reused mid-drain never produces two live executors for one channel.
 5166414            return _owner._executors.TryEnqueue(ChannelName, () => ProcessUnderCapturedContextAsync(messageValue)) switc
 2584415            {
 2584416                // Suppressed = a tombstoned channel with no registration left: the wait is gone and
 2584417                // the message would run against nobody (EnqueueAsync dropped these the same way).
 2584418                SerialExecutorRegistry.TryEnqueueOutcome.Accepted or SerialExecutorRegistry.TryEnqueueOutcome.Suppressed
 2582419                    => Task.CompletedTask,
 2420                _ => OnOverloadedAsync()
 2584421            };
 422        }
 423
 424        /// <summary>
 425        /// The overload outcome: the bounded per-correlation-id buffer is full and the next response
 426        /// cannot be admitted. Faults the wait with the explicit indeterminate contract (a terminal
 427        /// response may be queued or may be the one refused) and tears the subscription down so the
 428        /// flood stops here. Deliberately <see cref="CleanupOnceAsync"/> rather than the drain: the
 429        /// executor is full, and parking on a drain marker would block the subscriber's message
 430        /// loop — exactly the unbounded buffering this refuses. A full executor that is merely
 431        /// mid-retirement means cleanup already settled the task, and the message is a straggler.
 432        /// </summary>
 433        private async Task OnOverloadedAsync()
 434        {
 2435            var overload = new AsyncResponseIndeterminateDeliveryException(_correlationId, ChannelSerialExecutor.Default
 2436            Interlocked.Exchange(ref _overloaded, 1);
 2437            if (!_tcs.TrySetException(overload))
 438            {
 0439                if (_owner._logger.IsEnabled(LogLevel.Debug))
 0440                    _owner._logger.LogDebug("Dropped a late message on channel {Channel}: the wait for correlationId {Co
 0441                return;
 442            }
 443
 2444            _owner._logger.LogError(
 2445                "Wait for correlationId {CorrelationId} is overloaded: {Buffered} responses are queued behind its serial
 2446                _correlationId,
 2447                ChannelSerialExecutor.DefaultCapacity);
 2448            AsyncResponseDiagnostics.SetError(_activity, "overloaded", "The wait's bounded response buffer overflowed.")
 2449            AsyncResponseDiagnostics.RecordWaiterOverload("redis");
 2450            await CleanupOnceAsync().ConfigureAwait(false);
 2451        }
 452
 453        /// <summary>
 454        /// Restores the waiter's subscribe-time ExecutionContext (app AsyncLocals: trace, principal,
 455        /// logging scope) plus the correlation id before processing — the Redis subscriber callback
 456        /// runs on a foreign thread-pool thread that never had them.
 457        /// </summary>
 458        private Task ProcessUnderCapturedContextAsync(RedisValue messageValue)
 459        {
 460            async Task ProcessAsync()
 461            {
 2582462                using var correlationScope = AsyncResponseContext.PushCorrelationId(_correlationId);
 2582463                await ProcessMessageAsync(messageValue).ConfigureAwait(false);
 2582464            }
 465
 2582466            if (_capturedContext is null)
 2467                return ProcessAsync();
 468
 2580469            Task? task = null;
 5160470            ExecutionContext.Run(_capturedContext, _ => task = ProcessAsync(), null);
 2580471            return task!;
 472        }
 473
 474        /// <summary>Deserializes and handles a single incoming envelope, completes the TCS when terminal.</summary>
 475        private async Task ProcessMessageAsync(RedisValue messageValue)
 476        {
 2582477            if (Volatile.Read(ref _overloaded) != 0)
 478            {
 479                // Queued behind the overload fault: the wait is settled as indeterminate and the
 480                // subscription torn down, so running the predicate would spend user code — up to a
 481                // full executor's worth of it — on an outcome that cannot change. Only the overload
 482                // skips: a message admitted ahead of an ordinary terminal settlement still runs, as
 483                // the retirement drain expects.
 2048484                if (_owner._logger.IsEnabled(LogLevel.Debug))
 0485                    _owner._logger.LogDebug("Dropped a queued message on channel {Channel}: the wait for correlationId {
 2048486                return;
 487            }
 488
 534489            _owner._logger.LogDebug("Received message on channel {Channel}.", ChannelName);
 490
 534491            bool finished = false;
 492            try
 493            {
 494                // The delivered value is UTF-8 bytes; deserializing them directly avoids the
 495                // ToString() detour, which paid a payload-sized UTF-16 allocation plus a
 496                // transcode both ways on every message. Through JsonSafety, not the raw reader:
 497                // a parse failure lands in the catch below, which logs it AND hands it to the
 498                // waiter, and the reader's own message quotes the inbound body — property names
 499                // and dictionary keys straight off the wire (docs/security.md, "never logs a
 500                // message body"). Only the size and position survive.
 534501                var envelope = JsonSafety.SafeDeserialize((ReadOnlySpan<byte>)(byte[]?)messageValue, AsyncResponseEnvelo
 502
 526503                if (envelope == null)
 504                {
 6505                    _owner._logger.LogError("Failed to deserialize envelope for correlationId {CorrelationId}.", _correl
 506
 6507                    finished = true;
 6508                    var deserializationError = new JsonException($"Failed to deserialize envelope for correlationId {_co
 6509                    AsyncResponseDiagnostics.SetError(_activity, "deserialize_failure", deserializationError.Message);
 6510                    if (!_tcs.TrySetException(deserializationError))
 2511                        _owner._logger.LogWarning(deserializationError, "TaskCompletionSource already completed for corr
 512                }
 520513                else if (!AsyncResponseEnvelopeSchema.IsReadable(envelope.SchemaVersion))
 514                {
 6515                    finished = true;
 6516                    var schemaError = new InvalidOperationException(
 6517                        $"Response envelope for correlationId {_correlationId} has schema version {envelope.SchemaVersio
 6518                        $"which this build does not support (current: {AsyncResponseEnvelopeSchema.Current}).");
 6519                    AsyncResponseDiagnostics.SetError(_activity, "schema_mismatch", schemaError.Message);
 6520                    if (!_tcs.TrySetException(schemaError))
 2521                        _owner._logger.LogWarning(schemaError, "TaskCompletionSource already completed for correlationId
 522                }
 514523                else if (!envelope.Success)
 524                {
 7525                    finished = true;
 7526                    var remoteFailure = new Exception(envelope.ExceptionMessage ?? "Unknown error during asynchronous pr
 7527                    if (!string.IsNullOrEmpty(envelope.ExceptionStackTrace))
 528                    {
 529                        // Cap on receive too: the publish-side cap only bounds traces we emit, not what
 530                        // a remote we do not control can push at us.
 2531                        remoteFailure.Data["RemoteStackTrace"] = RemoteStackTrace.Cap(envelope.ExceptionStackTrace, _own
 532                    }
 533
 7534                    _owner._logger.LogWarning("Received error response for correlationId {CorrelationId}: {ErrorMessage}
 7535                    AsyncResponseDiagnostics.SetError(_activity, "remote_failure", remoteFailure.Message);
 7536                    if (!_tcs.TrySetException(remoteFailure))
 2537                        _owner._logger.LogWarning(remoteFailure, "TaskCompletionSource already completed for correlation
 538                }
 539                else
 540                {
 507541                    if (_owner._logger.IsEnabled(LogLevel.Debug))
 2542                        _owner._logger.LogDebug("Received response for correlationId {CorrelationId}.", _correlationId);
 543
 507544                    finished = await _completionPredicate(envelope.Payload!).ConfigureAwait(false);
 545
 507546                    if (finished && !_tcs.TrySetResult(envelope.Payload!))
 5547                        _owner._logger.LogWarning("TaskCompletionSource already completed for correlationId {Correlation
 548                }
 526549            }
 8550            catch (Exception ex)
 551            {
 8552                _owner._logger.LogError(ex, "Error processing message on channel {Channel} for correlationId {Correlatio
 553
 8554                finished = true;
 8555                AsyncResponseDiagnostics.SetError(_activity, ex);
 8556                if (!_tcs.TrySetException(ex))
 2557                    _owner._logger.LogWarning(ex, "TaskCompletionSource already completed for correlationId {Correlation
 8558            }
 559            finally
 560            {
 561                // Unsubscription also happens on dispose, but doing it immediately after the
 562                // terminal message releases resources sooner.
 534563                if (finished)
 413564                    await CleanupOnceAsync().ConfigureAwait(false);
 565            }
 2582566        }
 567
 568        /// <summary>
 569        /// Task-latched so EVERY caller completes only when the one real cleanup has finished —
 570        /// a fire-once flag alone would let a second caller (a disposing waiter racing the
 571        /// timeout) return before the task was settled.
 572        /// </summary>
 573        public ValueTask CleanupOnceAsync()
 574        {
 575            Task task;
 852576            lock (_cleanupGate)
 577            {
 852578                task = _cleanupTask ??= CleanupCoreAsync();
 852579            }
 580
 852581            return task.IsCompletedSuccessfully ? ValueTask.CompletedTask : new ValueTask(task);
 582        }
 583
 584        /// <summary>
 585        /// Dispose-path cleanup: DRAINS the per-channel serial executor before settling. A delivery
 586        /// may be mid <c>Until</c>-predicate holding a claimed terminal message; the marker work
 587        /// item completes only after that in-flight item finished, so by the time cleanup cancels,
 588        /// the task is either settled by the delivery or genuinely undelivered — never a
 589        /// cancellation stealing a consumed response. Must NOT be called from dispatch code (which
 590        /// runs ON the executor): the dispatch-triggered cleanup uses <see cref="CleanupOnceAsync"/>
 591        /// directly, its task already settled.
 592        /// <para>
 593        /// The drain is bounded by <c>DisposalDrainTimeout</c> — one budget covering marker
 594        /// ADMISSION too (a full bounded queue behind a wedged item blocks the enqueue itself). A
 595        /// lapsed budget must not fall back to the cleanup's cancel: the wedged delivery holds a
 596        /// message already consumed from the stream, and "canceled" would tell a re-attaching
 597        /// caller nothing was delivered. It faults the task with the explicit indeterminate
 598        /// contract instead, routing durable flows to a fresh idempotent restart. A
 599        /// tombstone-suppressed enqueue is the opposite case — the retired executor finished
 600        /// everything it ever admitted, so nothing is in flight and the plain cancel is truthful.
 601        /// </para>
 602        /// </summary>
 603        public async ValueTask DrainThenCleanupAsync(Exception? terminalIfUndelivered = null)
 604        {
 437605            if (Volatile.Read(ref _cleanupStarted) == 0 && ExecutorRegistered)
 606            {
 31607                var drainTimeout = _owner._options.DisposalDrainTimeout;
 31608                var drained = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
 609                try
 610                {
 31611                    using var budget = new CancellationTokenSource(drainTimeout);
 31612                    var accepted = await _owner._executors.EnqueueAsync(ChannelName, () =>
 31613                    {
 31614                        drained.TrySetResult();
 31615                        return Task.CompletedTask;
 31616                    }, budget.Token).ConfigureAwait(false);
 31617                    if (accepted)
 31618                        await drained.Task.WaitAsync(budget.Token).ConfigureAwait(false);
 30619                }
 1620                catch (Exception drainEx)
 621                {
 622                    // Budget lapse — or an unforeseen drain failure: either way the marker never
 623                    // ran, so an in-flight delivery cannot be ruled out (only accepted=false
 624                    // proves the executor finished everything). Settlement unproven means the
 625                    // cleanup's cancel below would be a false "nothing was delivered" — fault
 626                    // with the explicit indeterminate contract instead. A TrySetResult from the
 627                    // late-finishing dispatch loses against this and is dropped; its cleanup
 628                    // call is a no-op behind the latch.
 1629                    _owner._logger.LogWarning(
 1630                        "Disposal drain for correlationId {CorrelationId} did not prove settlement within {DrainTimeout}
 1631                        _correlationId, drainTimeout);
 1632                    AsyncResponseDiagnostics.SetError(_activity, "indeterminate_delivery", "Disposal drain did not prove
 1633                    if (drainEx is not OperationCanceledException)
 0634                        _owner._logger.LogDebug(drainEx, "Dispatch drain failed for channel {Channel}.", ChannelName);
 1635                    _tcs.TrySetException(new AsyncResponseIndeterminateDeliveryException(_correlationId, drainTimeout));
 1636                }
 31637            }
 638
 639            // Settle AFTER the drain, never before it. A delivery already inside the per-correlation
 640            // executor may hold a message the claim acked — the publisher was told "delivered", so
 641            // it exists nowhere else. Faulting first let a timeout beat that in-flight delivery and
 642            // report a consumed response as a timeout; TrySet loses here if the delivery won, which
 643            // is the whole point. (A lapsed drain budget has already faulted the task as
 644            // indeterminate above, and TrySet is a no-op behind it.)
 437645            if (terminalIfUndelivered is not null)
 8646                _tcs.TrySetException(terminalIfUndelivered);
 647
 437648            await CleanupOnceAsync().ConfigureAwait(false);
 437649        }
 650
 651        private async Task CleanupCoreAsync()
 652        {
 429653            Interlocked.Exchange(ref _cleanupStarted, 1);
 654
 655            // A waiter disposed before any terminal signal must not leave ResponseTask pending
 656            // forever for callers that hold it directly — the timeout dies with this cleanup, so
 657            // nothing else could ever complete the task. Cancellation is a no-op after a normal
 658            // completion, timeout, or fault (and after a delivery drained by DrainThenCleanupAsync).
 429659            _tcs.TrySetCanceled();
 660
 661            try
 662            {
 663                try
 664                {
 665                    // Delete the recovery state BEFORE unsubscribing. In the reverse order a publish
 666                    // landing in the window sees "no subscriber, state present" and fires a spurious
 667                    // recovery callback for a wait that already reached a terminal state. In this
 668                    // order the window shows a subscriber that drops the message — a late or duplicate
 669                    // terminal message is droppable; a resurrected recovery callback is not.
 429670                    await _owner._recoveryStateStore.TryDeleteAsync(_correlationId, Id).ConfigureAwait(false);
 427671                }
 2672                catch (Exception ex)
 673                {
 674                    // Best-effort: the state expires on its own, and a transient store failure must
 675                    // not skip the unsubscribe and executor teardown below.
 2676                    _owner._logger.LogError(ex, "Failed to delete recovery state for correlationId {CorrelationId}.", _c
 2677                }
 678
 679                try
 680                {
 681                    // Bounded like the drain: this latched core is what a disposing waiter awaits
 682                    // when terminal delivery started cleanup first, so an unbudgeted unsubscribe
 683                    // would let a wedged client library hold DisposeAsync hostage past
 684                    // DisposalDrainTimeout. The quiet wrapper logs its own failure — including
 685                    // one that completes AFTER this wait was abandoned, which previously died as
 686                    // a TaskScheduler.UnobservedTaskException nobody logged.
 429687                    if (Subscription is not null)
 427688                        await UnsubscribeQuietlyAsync(Subscription).WaitAsync(_owner._options.DisposalDrainTimeout).Conf
 429689                }
 0690                catch (TimeoutException)
 691                {
 0692                    _owner._logger.LogError(
 0693                        "Unsubscribe for channel {Channel} did not finish within {DisposalDrainTimeout}; abandoning the 
 0694                        ChannelName, _owner._options.DisposalDrainTimeout);
 0695                }
 696            }
 697            finally
 698            {
 699                // Purely local teardown runs no matter which network call above failed — the
 700                // cleanup latch is already set, so anything skipped here would leak until process
 701                // exit.
 429702                if (ExecutorRegistered)
 429703                    _owner._executors.OnSubscriptionRetired(ChannelName);
 704
 705                // Schedule the disposal on the thread pool; do not await directly to prevent
 706                // deadlocks with work currently running on the executor. Tracked so the channel's
 707                // DisposeAsync can join it at host shutdown.
 429708                _owner.TrackRetirement(Task.Run(async () =>
 429709                {
 429710                    try
 429711                    {
 429712                        await _owner._executors.RemoveAsync(ChannelName).ConfigureAwait(false);
 429713                    }
 0714                    catch (Exception ex)
 429715                    {
 0716                        _owner._logger.LogError(ex, "Failed to retire the executor for channel {Channel}.", ChannelName)
 0717                    }
 858718                }));
 719
 429720                await _timeoutRegistration.DisposeAsync().ConfigureAwait(false);
 429721                _cancellationTokenSource.Dispose();
 429722                _activity?.Dispose();
 723            }
 429724        }
 725
 726        /// <summary>
 727        /// Never faults: the unsubscribe outcome is logged HERE, so a teardown outliving the
 728        /// bounded wait above still records its failure instead of surfacing as an unobserved
 729        /// task exception.
 730        /// </summary>
 731        public async Task UnsubscribeQuietlyAsync(IRedisChannelSubscription liveSubscription)
 732        {
 733            try
 734            {
 427735                await liveSubscription.DisposeAsync().ConfigureAwait(false);
 425736                _owner._logger.LogDebug("Unsubscribed from channel {Channel}.", ChannelName);
 425737            }
 2738            catch (Exception ex)
 739            {
 2740                _owner._logger.LogError(ex, "Error during unsubscribe-once for channel {Channel}.", ChannelName);
 2741            }
 427742        }
 743    }
 744
 745    // ---------------------------------------------------------------------------------------
 746    // IAsyncResponsePublisher
 747
 748    /// <inheritdoc/>
 749    public Task SetResponse<T>(T response, string correlationId, CancellationToken cancellationToken = default) where T 
 513750        => SetResponseCore(response, correlationId, cancellationToken);
 751
 752    Task IRawAsyncResponsePublisher.SetRawResponse(object? response, string correlationId, CancellationToken cancellatio
 2753        => SetResponseCore(response, correlationId, cancellationToken);
 754
 755    Task IRawAsyncResponsePublisher.SetRawResponseJson(string responseJson, string correlationId, CancellationToken canc
 44756        => SetRawResponseJsonCore(responseJson, correlationId, cancellationToken);
 757
 758    // Intentionally duplicated with SetRawResponseJsonCore: this publish method is a latency hot
 759    // path, and earlier shared helper/delegate refactors regressed throughput in benchmarks.
 760    // Keep the typed Redis path inline unless a benchmark run proves a refactor is free.
 761    /// <summary>
 762    /// Retires the correlation id's serial executor after a recovery-routed publish, bounded by
 763    /// <c>DisposalDrainTimeout</c>. The registry's own removal joins the executor's retirement,
 764    /// and that retirement can be draining a work item wedged in a user <c>Until</c> predicate —
 765    /// so an unbounded join here stalled the ingress consumer thread per late/duplicate response
 766    /// for the registry's 30 s + 30 s defaults, with no configured budget applying. The
 767    /// retirement itself continues in the background once the wait lapses.
 768    /// </summary>
 769    private async ValueTask RetireExecutorBoundedAsync(string channel)
 770    {
 771        try
 772        {
 59773            await _executors.RemoveAsync(channel).AsTask().WaitAsync(_options.DisposalDrainTimeout).ConfigureAwait(false
 53774        }
 6775        catch (TimeoutException)
 776        {
 6777            _logger.LogWarning(
 6778                "Retiring the serial executor for channel {Channel} did not complete within DisposalDrainTimeout ({Dispo
 6779                channel,
 6780                _options.DisposalDrainTimeout);
 6781        }
 59782    }
 783
 784    private async Task SetResponseCore<T>(T response, string correlationId, CancellationToken cancellationToken)
 785    {
 515786        using var activity = AsyncResponseDiagnostics.StartActivity("asyncresponse.set_response", ActivityKind.Producer)
 515787        activity?.SetTag("asyncresponse.channel", "redis");
 515788        AsyncResponseDiagnostics.SetPayloadType(activity, typeof(T));
 789
 790        // When no correlation id is provided, fall back to the ambient context.
 515791        AsyncResponseDiagnostics.SetCorrelationId(activity, correlationId);
 792
 515793        if (CorrelationIdGuard.IsUnpublishable(correlationId, _logger, activity, "the response"))
 4794            return;
 795
 508796        var channel = _keys.Channel(correlationId);
 797        try
 798        {
 508799            var envelope = new AsyncResponseEnvelope<T>
 508800            {
 508801                Success = true,
 508802                Payload = response
 508803            };
 508804            var json = AsyncResponseEnvelopeJson.Serialize(envelope);
 508805            long numSubscribers = await _subscriber.PublishAsync(channel, json).ConfigureAwait(false);
 506806            activity?.SetTag("asyncresponse.subscribers", numSubscribers);
 807
 506808            if (numSubscribers == 0)
 809            {
 810                // Nobody was listening (the waiter died, e.g. with a redeploy): hand the response
 811                // over to the lost-subscriber dispatcher, which asks the payload whether to resume
 812                // the flow or fail it, and invokes the matching callback.
 31813                var dispatchResult = await _lostSubscriberDispatcher
 31814                    .DispatchLostResponses(
 31815                        _recoveryStateStore,
 31816                        correlationId,
 31817                        response,
 31818                        channel.ToString()!,
 31819                        cancellationToken,
 31820                        hasLiveSubscriber: () => HasLiveSubscriberAsync(correlationId, cancellationToken))
 31821                    .ConfigureAwait(false);
 25822                if (dispatchResult.RetryLive)
 823                {
 824                    // A waiter subscribed between the publish and the recovery-state read —
 825                    // re-publish live instead of consuming its registration; only a second miss
 826                    // consumes it.
 6827                    numSubscribers = await _subscriber.PublishAsync(channel, json).ConfigureAwait(false);
 6828                    activity?.SetTag("asyncresponse.subscribers", numSubscribers);
 6829                    if (numSubscribers > 0)
 2830                        return;
 831
 4832                    dispatchResult = await _lostSubscriberDispatcher
 4833                        .DispatchLostResponses(
 4834                            _recoveryStateStore,
 4835                            correlationId,
 4836                            response,
 4837                            channel.ToString()!,
 4838                            cancellationToken,
 4839                            hasLiveSubscriber: () => HasLiveSubscriberAsync(correlationId, cancellationToken))
 4840                        .ConfigureAwait(false);
 4841                    if (dispatchResult.RetryLive)
 842                    {
 843                        // Second contradiction: delivery keeps reporting no responders while the
 844                        // probe keeps reporting a live subscriber (interest not yet visible
 845                        // server-side, or a stale heartbeat). Consuming registrations on this
 846                        // evidence would strip a live waiter of its recovery arm — leave all state
 847                        // intact and surface the non-delivery to the caller, whose retry/redelivery
 848                        // machinery re-attempts once the subscription is visible (bounded by the
 849                        // heartbeat's liveness expiry, after which normal recovery takes over).
 850                        // Returning here instead would silently drop the payload.
 2851                        _logger.LogWarning(
 2852                            "Delivery for correlationId {CorrelationId} found no subscribers twice while the liveness pr
 2853                            correlationId);
 2854                        activity?.SetTag("asyncresponse.recovery.liveness_contradiction", true);
 2855                        throw new InvalidOperationException(
 2856                            $"Redis delivery for correlationId '{correlationId}' found no subscribers twice while the li
 2857                            "reporting one; the payload was not delivered and recovery registrations were left intact. R
 2858                            "once the waiter's subscription is visible to the publishing endpoint.");
 859                    }
 860                }
 861
 21862                AsyncResponseDiagnostics.SetLostSubscriberRoute(activity, dispatchResult.Action, dispatchResult.RouteMix
 21863                AsyncResponseDiagnostics.RecordLostSubscriber("response", dispatchResult.Action, dispatchResult.Callback
 21864                activity?.SetTag("asyncresponse.recovery.callback_invoked", dispatchResult.CallbackInvoked);
 865
 21866                await RetireExecutorBoundedAsync(channel.ToString()!).ConfigureAwait(false);
 867            }
 868            else
 869            {
 475870                if (_logger.IsEnabled(LogLevel.Debug))
 4871                    _logger.LogDebug("Published response for correlationId {CorrelationId} on channel {Channel}. Payload
 872            }
 496873        }
 10874        catch (Exception ex)
 875        {
 10876            _logger.LogError(ex, "Failed to publish response for correlationId {CorrelationId} on channel {Channel}.", c
 10877            AsyncResponseDiagnostics.SetError(activity, ex);
 10878            throw;
 879        }
 502880    }
 881
 882    // Intentionally duplicated with SetResponseCore: raw ingress uses pre-serialized payload JSON
 883    // and a different lost-subscriber materialization path, so avoiding shared indirection matters.
 884    private async Task SetRawResponseJsonCore(string responseJson, string correlationId, CancellationToken cancellationT
 885    {
 44886        using var activity = AsyncResponseDiagnostics.StartActivity("asyncresponse.ingress.raw_response", ActivityKind.P
 44887        activity?.SetTag("asyncresponse.channel", "redis");
 888
 44889        AsyncResponseDiagnostics.SetCorrelationId(activity, correlationId);
 890
 44891        if (CorrelationIdGuard.IsUnpublishable(correlationId, _logger, activity, "the raw response", dropContractViolati
 4892            return;
 893
 40894        var channel = _keys.Channel(correlationId);
 895        try
 896        {
 40897            var json = SerializeRawSuccessEnvelope(responseJson);
 38898            long numSubscribers = await _subscriber.PublishAsync(channel, json).ConfigureAwait(false);
 36899            activity?.SetTag("asyncresponse.subscribers", numSubscribers);
 900
 36901            if (numSubscribers == 0)
 902            {
 31903                var response = new RawJsonResponse(responseJson).DeserializeUntyped();
 904
 31905                var dispatchResult = await _lostSubscriberDispatcher
 31906                    .DispatchLostResponses(
 31907                        _recoveryStateStore,
 31908                        correlationId,
 31909                        response,
 31910                        channel.ToString()!,
 31911                        cancellationToken,
 31912                        hasLiveSubscriber: () => HasLiveSubscriberAsync(correlationId, cancellationToken))
 31913                    .ConfigureAwait(false);
 29914                if (dispatchResult.RetryLive)
 915                {
 916                    // A waiter subscribed between the publish and the recovery-state read —
 917                    // re-publish live instead of consuming its registration; only a second miss
 918                    // consumes it.
 6919                    numSubscribers = await _subscriber.PublishAsync(channel, json).ConfigureAwait(false);
 6920                    activity?.SetTag("asyncresponse.subscribers", numSubscribers);
 6921                    if (numSubscribers > 0)
 2922                        return;
 923
 4924                    dispatchResult = await _lostSubscriberDispatcher
 4925                        .DispatchLostResponses(
 4926                            _recoveryStateStore,
 4927                            correlationId,
 4928                            response,
 4929                            channel.ToString()!,
 4930                            cancellationToken,
 4931                            hasLiveSubscriber: () => HasLiveSubscriberAsync(correlationId, cancellationToken))
 4932                        .ConfigureAwait(false);
 4933                    if (dispatchResult.RetryLive)
 934                    {
 935                        // Second contradiction: delivery keeps reporting no responders while the
 936                        // probe keeps reporting a live subscriber (interest not yet visible
 937                        // server-side, or a stale heartbeat). Consuming registrations on this
 938                        // evidence would strip a live waiter of its recovery arm — leave all state
 939                        // intact and surface the non-delivery to the caller, whose retry/redelivery
 940                        // machinery re-attempts once the subscription is visible (bounded by the
 941                        // heartbeat's liveness expiry, after which normal recovery takes over).
 942                        // Returning here instead would silently drop the payload.
 2943                        _logger.LogWarning(
 2944                            "Delivery for correlationId {CorrelationId} found no subscribers twice while the liveness pr
 2945                            correlationId);
 2946                        activity?.SetTag("asyncresponse.recovery.liveness_contradiction", true);
 2947                        throw new InvalidOperationException(
 2948                            $"Redis delivery for correlationId '{correlationId}' found no subscribers twice while the li
 2949                            "reporting one; the payload was not delivered and recovery registrations were left intact. R
 2950                            "once the waiter's subscription is visible to the publishing endpoint.");
 951                    }
 952                }
 953
 25954                AsyncResponseDiagnostics.SetLostSubscriberRoute(activity, dispatchResult.Action, dispatchResult.RouteMix
 25955                AsyncResponseDiagnostics.RecordLostSubscriber("response", dispatchResult.Action, dispatchResult.Callback
 25956                activity?.SetTag("asyncresponse.recovery.callback_invoked", dispatchResult.CallbackInvoked);
 957
 25958                await RetireExecutorBoundedAsync(channel.ToString()!).ConfigureAwait(false);
 25959            }
 960            else
 961            {
 5962                if (_logger.IsEnabled(LogLevel.Debug))
 2963                    _logger.LogDebug("Published raw response for correlationId {CorrelationId} on channel {Channel}. Sub
 964            }
 30965        }
 8966        catch (Exception ex)
 967        {
 8968            _logger.LogError(ex, "Failed to publish raw response for correlationId {CorrelationId} on channel {Channel}.
 8969            AsyncResponseDiagnostics.SetError(activity, ex);
 8970            throw;
 971        }
 36972    }
 973
 974    /// <inheritdoc/>
 975    public async Task SetException(Exception exception, string correlationId, CancellationToken cancellationToken = defa
 976    {
 44977        ArgumentNullException.ThrowIfNull(exception);
 978
 42979        using var activity = AsyncResponseDiagnostics.StartActivity("asyncresponse.set_exception", ActivityKind.Producer
 42980        activity?.SetTag("asyncresponse.channel", "redis");
 42981        activity?.SetTag("asyncresponse.exception_type", exception.GetType().FullName ?? exception.GetType().Name);
 982
 42983        AsyncResponseDiagnostics.SetCorrelationId(activity, correlationId);
 984
 42985        if (CorrelationIdGuard.IsUnpublishable(correlationId, _logger, activity, "the exception", exception))
 3986            return;
 987
 38988        var channel = _keys.Channel(correlationId);
 989        try
 990        {
 38991            var envelope = new AsyncResponseEnvelope<object>
 38992            {
 38993                Success = false,
 38994                ExceptionMessage = exception.Message,
 38995                ExceptionStackTrace = RemoteStackTrace.ForWire(exception.StackTrace, _options.IncludeRemoteStackTrace, _
 38996                Payload = null
 38997            };
 38998            var json = AsyncResponseEnvelopeJson.Serialize(envelope);
 38999            long numSubscribers = await _subscriber.PublishAsync(channel, json).ConfigureAwait(false);
 361000            activity?.SetTag("asyncresponse.subscribers", numSubscribers);
 1001
 361002            if (numSubscribers == 0)
 1003            {
 1004                // Nobody was listening: exception envelopes always go to the failure callback.
 231005                var dispatchResult = await _lostSubscriberDispatcher
 231006                    .DispatchLostExceptions(
 231007                        _recoveryStateStore,
 231008                        correlationId,
 231009                        exception,
 231010                        channel.ToString()!,
 231011                        cancellationToken,
 231012                        hasLiveSubscriber: () => HasLiveSubscriberAsync(correlationId, cancellationToken))
 231013                    .ConfigureAwait(false);
 171014                if (dispatchResult.RetryLive)
 1015                {
 1016                    // A waiter subscribed between the publish and the recovery-state read —
 1017                    // re-publish live instead of consuming its registration; only a second miss
 1018                    // consumes it.
 61019                    numSubscribers = await _subscriber.PublishAsync(channel, json).ConfigureAwait(false);
 61020                    activity?.SetTag("asyncresponse.subscribers", numSubscribers);
 61021                    if (numSubscribers > 0)
 21022                        return;
 1023
 41024                    dispatchResult = await _lostSubscriberDispatcher
 41025                        .DispatchLostExceptions(
 41026                            _recoveryStateStore,
 41027                            correlationId,
 41028                            exception,
 41029                            channel.ToString()!,
 41030                            cancellationToken,
 41031                            hasLiveSubscriber: () => HasLiveSubscriberAsync(correlationId, cancellationToken))
 41032                        .ConfigureAwait(false);
 41033                    if (dispatchResult.RetryLive)
 1034                    {
 1035                        // Second contradiction: delivery keeps reporting no responders while the
 1036                        // probe keeps reporting a live subscriber (interest not yet visible
 1037                        // server-side, or a stale heartbeat). Consuming registrations on this
 1038                        // evidence would strip a live waiter of its recovery arm — leave all state
 1039                        // intact and surface the non-delivery to the caller, whose retry/redelivery
 1040                        // machinery re-attempts once the subscription is visible (bounded by the
 1041                        // heartbeat's liveness expiry, after which normal recovery takes over).
 1042                        // Returning here instead would silently drop the payload.
 21043                        _logger.LogWarning(
 21044                            "Delivery for correlationId {CorrelationId} found no subscribers twice while the liveness pr
 21045                            correlationId);
 21046                        activity?.SetTag("asyncresponse.recovery.liveness_contradiction", true);
 21047                        throw new InvalidOperationException(
 21048                            $"Redis delivery for correlationId '{correlationId}' found no subscribers twice while the li
 21049                            "reporting one; the payload was not delivered and recovery registrations were left intact. R
 21050                            "once the waiter's subscription is visible to the publishing endpoint.");
 1051                    }
 1052                }
 1053
 131054                activity?.SetTag("asyncresponse.recovery.callback_invoked", dispatchResult.CallbackInvoked);
 131055                AsyncResponseDiagnostics.RecordLostSubscriber("exception", action: RecoveryAction.Fail, dispatchResult.C
 1056
 131057                await RetireExecutorBoundedAsync(channel.ToString()!).ConfigureAwait(false);
 1058            }
 131059            else if (_logger.IsEnabled(LogLevel.Debug))
 1060            {
 61061                _logger.LogDebug("Published exception response for correlationId {CorrelationId} on channel {Channel}. S
 1062            }
 261063        }
 101064        catch (Exception ex)
 1065        {
 101066            _logger.LogError(ex, "Failed to publish exception response for correlationId {CorrelationId} on channel {Cha
 101067            AsyncResponseDiagnostics.SetError(activity, ex);
 101068            throw;
 1069        }
 311070    }
 1071
 1072    // ---------------------------------------------------------------------------------------
 1073    // IActiveSubscriberProbe
 1074
 1075    /// <inheritdoc/>
 1076    /// <remarks>
 1077    /// Returns the live subscriber count, or a negative value when liveness could not be
 1078    /// established. Zero is reported only when every node that could hold the subscription
 1079    /// answered: the channels are key-routed, so the subscription lives on the single slot owner,
 1080    /// and a zero collected while that node was unreachable says nothing about the waiter.
 1081    /// </remarks>
 1082    public async ValueTask<long> CountActiveSubscribersAsync(string correlationId, CancellationToken cancellationToken =
 1083    {
 1261084        if (string.IsNullOrWhiteSpace(correlationId))
 41085            return 0L;
 1086
 1221087        var channel = _keys.Channel(correlationId);
 1088
 1089        // Subscriptions live on whichever node the client subscribed through, so the live count is
 1090        // the maximum reported across all connected endpoints. The async server call keeps large
 1091        // watchdog probe sweeps off blocking thread-pool waits, and the per-endpoint token check
 1092        // lets a shutdown abort the sweep between probes.
 1221093        long subscribers = 0;
 1221094        var answeredEveryPrimary = true;
 1221095        var primaryAnswered = false;
 5141096        foreach (var endPoint in _multiplexer.GetEndPoints())
 1097        {
 1361098            cancellationToken.ThrowIfCancellationRequested();
 1099
 1341100            var server = _multiplexer.GetServer(endPoint);
 1101
 1102            // PUBSUB NUMSUB is node-local and the response channels are key-routed, so the
 1103            // subscription sits on the ONE node that owns the channel key's slot — a primary. A
 1104            // node that has never connected reports the default (not a replica), which counts it
 1105            // as a primary here: the conservative direction, since the unknown node may be the
 1106            // very owner. Replicas are asked too (a positive answer is proof wherever it comes
 1107            // from) but never decide a zero.
 1341108            var isPrimary = !server.IsReplica;
 1341109            if (!server.IsConnected)
 1110            {
 101111                answeredEveryPrimary &= !isPrimary;
 101112                continue;
 1113            }
 1114
 1115            try
 1116            {
 1241117                subscribers = Math.Max(subscribers, await server.SubscriptionSubscriberCountAsync(channel).ConfigureAwai
 1101118                primaryAnswered |= isPrimary;
 1101119            }
 141120            catch (Exception ex) when (ex is not OperationCanceledException)
 1121            {
 141122                answeredEveryPrimary &= !isPrimary;
 141123                _logger.LogDebug(ex, "Failed to read subscriber count for channel {Channel}.", channel.ToString()!);
 141124            }
 1125        }
 1126
 1127        // A count above zero is proof of a live waiter wherever it was read. A zero is only the
 1128        // absence of one on the nodes that ANSWERED: skipping the slot owner and returning the
 1129        // siblings' node-local zeros asserted "definitively no live waiter" for a waiter that was
 1130        // subscribed all along — consuming its recovery registration (a double resume) or
 1131        // dropping its response. Negative = "could not be probed", the watchdog's and the
 1132        // snapshot-race re-check's unknown-liveness contract.
 1201133        if (subscribers > 0)
 291134            return subscribers;
 1135
 911136        return primaryAnswered && answeredEveryPrimary ? 0L : -1L;
 1241137    }
 1138
 1139    /// <summary>
 1140    /// Re-probes waiter liveness for the lost-subscriber dispatcher's snapshot-race re-check,
 1141    /// using the same PUBSUB NUMSUB-based probe the watchdog uses. An unprobeable result THROWS
 1142    /// instead of reading as "no live waiter", so the failure propagates to the publisher's catch
 1143    /// and the publish retries rather than consuming a live waiter's recovery registration
 1144    /// (parity with the DB channels, whose re-check calls the store directly).
 1145    /// </summary>
 1146    private async ValueTask<bool> HasLiveSubscriberAsync(string correlationId, CancellationToken cancellationToken)
 1147    {
 971148        var subscribers = await CountActiveSubscribersAsync(correlationId, cancellationToken).ConfigureAwait(false);
 971149        if (subscribers < 0)
 1150        {
 61151            throw new InvalidOperationException(
 61152                $"Redis subscriber liveness for correlationId '{correlationId}' could not be probed on any connected end
 1153        }
 1154
 911155        return subscribers > 0;
 911156    }
 1157
 1158    private static string SerializeRawSuccessEnvelope(string payloadJson)
 1159    {
 401160        JsonSafety.ThrowIfClearlyNotJson(payloadJson);
 1161
 381162        var buffer = new ArrayBufferWriter<byte>();
 381163        using (var writer = new Utf8JsonWriter(buffer))
 1164        {
 381165            writer.WriteStartObject();
 381166            writer.WriteNumber("SchemaVersion", AsyncResponseEnvelopeSchema.Current);
 381167            writer.WriteBoolean("Success", true);
 381168            writer.WritePropertyName("Payload");
 381169            writer.WriteRawValue(payloadJson);
 381170            writer.WriteNull("ExceptionMessage");
 381171            writer.WriteNull("ExceptionStackTrace");
 381172            writer.WriteEndObject();
 381173        }
 1174
 381175        return Encoding.UTF8.GetString(buffer.WrittenSpan);
 1176    }
 1177}
Generated by: ReportGenerator 5.5.11.0
09/22/2026 - 09:41:04
GitHub | reportgenerator.io

Methods/Properties

.ctor(Microsoft.Extensions.DependencyInjection.IServiceScopeFactory,StackExchange.Redis.IConnectionMultiplexer,AsyncResponse.IRecoveryStateStore,Microsoft.Extensions.Options.IOptions`1<AsyncResponse.Channels.Redis.RedisAsyncResponseOptions>,AsyncResponse.AsyncResponseContextPropagation,Microsoft.Extensions.Logging.ILogger`1<AsyncResponse.Channels.Redis.RedisAsyncResponseChannel>,AsyncResponse.Channels.Redis.IRedisChannelSubscriber,System.TimeProvider)
TrackRetirement(System.Threading.Tasks.Task)
DisposeAsync()
CreateResponseWaiter(System.String,System.Func`2<T,System.Threading.Tasks.ValueTask`1<System.Boolean>>,System.Nullable`1<System.TimeSpan>)
CreateRecoverableResponseWaiter(System.String,AsyncResponse.ReflectionCallDto,AsyncResponse.ReflectionCallDto,System.Func`2<T,System.Threading.Tasks.ValueTask`1<System.Boolean>>,System.Nullable`1<System.TimeSpan>)
CreateResponseWaiterCore()
.ctor(AsyncResponse.Channels.Redis.RedisAsyncResponseChannel,System.String,StackExchange.Redis.RedisChannel,System.Guid,System.Func`2<T,System.Threading.Tasks.ValueTask`1<System.Boolean>>,System.Threading.ExecutionContext,System.Diagnostics.Activity)
get_Id()
get_ChannelName()
get_ResponseTask()
get_CleanupStarted()
get_Subscription()
get_ExecutorRegistered()
RegisterTimeoutCallback()
ArmTimeout(System.TimeSpan)
OnTimeout()
HandleMessageAsync(StackExchange.Redis.RedisChannel,StackExchange.Redis.RedisValue)
OnOverloadedAsync()
ProcessAsync()
ProcessUnderCapturedContextAsync(StackExchange.Redis.RedisValue)
ProcessMessageAsync()
CleanupOnceAsync()
DrainThenCleanupAsync()
CleanupCoreAsync()
<CleanupCoreAsync()
UnsubscribeQuietlyAsync()
SetResponse(T,System.String,System.Threading.CancellationToken)
AsyncResponse.IRawAsyncResponsePublisher.SetRawResponse(System.Object,System.String,System.Threading.CancellationToken)
AsyncResponse.IRawAsyncResponsePublisher.SetRawResponseJson(System.String,System.String,System.Threading.CancellationToken)
RetireExecutorBoundedAsync()
SetResponseCore()
SetRawResponseJsonCore()
SetException()
CountActiveSubscribersAsync()
HasLiveSubscriberAsync()
SerializeRawSuccessEnvelope(System.String)