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

Information
Class: AsyncResponse.Transports.Redis.AwaitingRedisMessageDispatcher
Assembly: AsyncResponse.Transports.Redis
File(s): /_/src/Transports/AsyncResponse.Transports.Redis/RedisMessageDispatcher.cs
Line coverage
100%
Covered lines: 43
Uncovered lines: 0
Coverable lines: 43
Total lines: 713
Line coverage: 100%
Branch coverage
100%
Covered branches: 2
Total branches: 2
Branch coverage: 100%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

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

File(s)

/_/src/Transports/AsyncResponse.Transports.Redis/RedisMessageDispatcher.cs

#LineLine coverage
 1using Microsoft.Extensions.Logging;
 2using StackExchange.Redis;
 3using System.Diagnostics;
 4using System.Threading.Channels;
 5
 6namespace AsyncResponse.Transports.Redis;
 7
 8internal enum RedisSubscriberRole
 9{
 10    Worker,
 11    ResponseIngress
 12}
 13
 14internal enum RedisDispatchOutcome
 15{
 16    /// <summary>The entry was handled, ACKed, or dead-lettered. Counts as progress for the poll loop.</summary>
 17    Processed,
 18
 19    /// <summary>The entry could not be accepted right now (background queue full) and was left pending for retry.</summ
 20    Deferred
 21}
 22
 23internal sealed record RedisStreamDelivery(
 24    RedisKey Stream,
 25    RedisValue ConsumerGroup,
 26    RedisValue MessageId,
 27    string Payload,
 28    string? CorrelationId,
 29    int Attempt,
 30    StreamEntry Entry);
 31
 32internal abstract class RedisMessageDispatcher : IAsyncDisposable
 33{
 34    private readonly Func<RedisStreamDelivery, CancellationToken, Task> _handler;
 35    private readonly RedisSubscriberOptions _subscriberOptions;
 36    private readonly IRedisStreamDatabase _database;
 37    private readonly RedisTransportKeySchema _keys;
 38    private readonly string _stream;
 39    private readonly string _consumerGroup;
 40    private readonly RedisSubscriberRole _role;
 41
 42    /// <summary>Runs the RedisMessageDispatcher operation.</summary>
 43    protected RedisMessageDispatcher(
 44        Func<RedisStreamDelivery, CancellationToken, Task> handler,
 45        IRedisStreamDatabase database,
 46        RedisAsyncResponseTransportOptions transportOptions,
 47        RedisSubscriberOptions subscriberOptions,
 48        ILogger logger,
 49        RedisKey stream,
 50        RedisValue consumerGroup,
 51        RedisSubscriberRole role)
 52    {
 53        _handler = handler;
 54        _database = database;
 55        TransportOptions = transportOptions;
 56        _subscriberOptions = subscriberOptions;
 57        _keys = new RedisTransportKeySchema(transportOptions);
 58        Logger = logger;
 59        _stream = stream.ToString();
 60        _consumerGroup = consumerGroup.ToString();
 61        _role = role;
 62    }
 63
 64    protected RedisAsyncResponseTransportOptions TransportOptions { get; }
 65    protected ILogger Logger { get; }
 66
 67    protected int MaxDeliveryAttempts => _subscriberOptions.MaxDeliveryAttempts;
 68
 69    /// <summary>Creates the configured dispatcher.</summary>
 70    public static RedisMessageDispatcher Create(
 71        Func<RedisStreamDelivery, CancellationToken, Task> handler,
 72        IRedisStreamDatabase database,
 73        RedisAsyncResponseTransportOptions transportOptions,
 74        RedisSubscriberOptions subscriberOptions,
 75        ILogger logger,
 76        RedisKey stream,
 77        RedisValue consumerGroup,
 78        RedisSubscriberRole role)
 79    {
 80        ValidateOptions(transportOptions, subscriberOptions, role);
 81
 82        if (subscriberOptions.AckMode is RedisAckMode.AckAfterEnqueue)
 83        {
 84            return new QueuedRedisMessageDispatcher(
 85                handler,
 86                database,
 87                transportOptions,
 88                subscriberOptions,
 89                logger,
 90                stream,
 91                consumerGroup,
 92                role);
 93        }
 94
 95        return new AwaitingRedisMessageDispatcher(
 96            handler,
 97            database,
 98            transportOptions,
 99            subscriberOptions,
 100            logger,
 101            stream,
 102            consumerGroup,
 103            role);
 104    }
 105
 106    /// <summary>Validates the supplied options.</summary>
 107    public static void ValidateOptions(
 108        RedisAsyncResponseTransportOptions transportOptions,
 109        RedisSubscriberOptions subscriberOptions,
 110        RedisSubscriberRole role)
 111    {
 112        RedisTransportOptionsValidator.ValidateCommon(transportOptions);
 113
 114        var optionPath = role is RedisSubscriberRole.Worker
 115            ? $"{nameof(RedisAsyncResponseTransportOptions)}.{nameof(RedisAsyncResponseTransportOptions.WorkerSubscriber
 116            : $"{nameof(RedisAsyncResponseTransportOptions)}.{nameof(RedisAsyncResponseTransportOptions.ResponseSubscrib
 117
 118        if (subscriberOptions.BatchSize <= 0)
 119            throw new InvalidOperationException($"{optionPath}.{nameof(RedisSubscriberOptions.BatchSize)} must be positi
 120        // EmptyPollDelay arms the idle Task.Delay (timer ceiling). PendingMessageMinIdleTime is
 121        // the server-side XAUTOCLAIM min-idle in milliseconds, but it ALSO arms the in-process
 122        // idle-reset heartbeat's Task.Delay at one third of its value, so its real sink is the
 123        // timer ceiling too — under the persistence bound a legal 200-day value passed validation
 124        // and then killed every batch with ArgumentOutOfRangeException from the heartbeat's delay.
 125        // PendingClaimInterval is a "now + interval" schedule stamp and keeps the persistence bound.
 126        AsyncResponseChannelOptions.EnsureTimerBacked(subscriberOptions.EmptyPollDelay, optionPath, nameof(RedisSubscrib
 127        AsyncResponseChannelOptions.EnsureTimerBacked(subscriberOptions.PendingMessageMinIdleTime, optionPath, nameof(Re
 128        AsyncResponseChannelOptions.EnsurePersistedTtl(subscriberOptions.PendingClaimInterval, optionPath, nameof(RedisS
 129        if (subscriberOptions.PendingClaimBatchSize <= 0)
 130            throw new InvalidOperationException($"{optionPath}.{nameof(RedisSubscriberOptions.PendingClaimBatchSize)} mu
 131        if (subscriberOptions.MaxDeliveryAttempts < 0)
 132            throw new InvalidOperationException($"{optionPath}.{nameof(RedisSubscriberOptions.MaxDeliveryAttempts)} cann
 133
 134        switch (subscriberOptions.AckMode)
 135        {
 136            case RedisAckMode.AckAfterHandlerCompletes:
 137                return;
 138
 139            case RedisAckMode.AckAfterEnqueue:
 140                if (subscriberOptions.BackgroundWorkerCount <= 0)
 141                {
 142                    throw new InvalidOperationException(
 143                        $"{optionPath}.{nameof(RedisSubscriberOptions.BackgroundWorkerCount)} must be explicitly configu
 144                        $"when {nameof(RedisSubscriberOptions.AckMode)} is {nameof(RedisAckMode.AckAfterEnqueue)}.");
 145                }
 146
 147                if (subscriberOptions.BackgroundQueueCapacity <= 0)
 148                {
 149                    throw new InvalidOperationException(
 150                        $"{optionPath}.{nameof(RedisSubscriberOptions.BackgroundQueueCapacity)} must be explicitly confi
 151                        $"when {nameof(RedisSubscriberOptions.AckMode)} is {nameof(RedisAckMode.AckAfterEnqueue)}.");
 152                }
 153
 154                AsyncResponseChannelOptions.EnsureTimerBacked(subscriberOptions.BackgroundDrainTimeout, optionPath, name
 155
 156                // Redis subscribers spend only the background drain at shutdown; the read loop
 157                // stops with the host token and the multiplexer teardown is not separately bounded.
 158                ShutdownBudgetValidator.Validate(
 159                    "Redis",
 160                    $"{nameof(RedisAsyncResponseTransportOptions)}.{nameof(RedisAsyncResponseTransportOptions.HostShutdo
 161                    transportOptions.HostShutdownTimeout,
 162                    ($"{optionPath}.{nameof(RedisSubscriberOptions.BackgroundDrainTimeout)}", subscriberOptions.Backgrou
 163
 164                return;
 165
 166            default:
 167                throw new InvalidOperationException(
 168                    $"{optionPath}.{nameof(RedisSubscriberOptions.AckMode)} has unsupported value '{subscriberOptions.Ac
 169        }
 170    }
 171
 172    /// <summary>Handles the delivered message.</summary>
 173    public abstract Task<RedisDispatchOutcome> HandleAsync(
 174        RedisStreamDelivery delivery,
 175        CancellationToken subscriberCancellationToken);
 176
 177    /// <summary>
 178    /// Whether the dispatcher can accept more deliveries right now. Awaiting dispatchers always can
 179    /// (handlers run inline); the queued dispatcher returns <c>false</c> while its bounded queue is
 180    /// saturated so the subscriber stops pulling new entries into the pending-entry list instead of
 181    /// busy-reading and rejecting them.
 182    /// </summary>
 183    public virtual bool CanAcceptMore => true;
 184
 185    /// <summary>
 186    /// How many entries the dispatcher can take right now without deferring any (ASB/SQS parity);
 187    /// unbounded for the awaiting dispatcher. The subscriber clamps every read and claim to it.
 188    /// </summary>
 189    public virtual int FreeCapacity => int.MaxValue;
 190
 191    /// <summary>Releases resources held by this instance.</summary>
 192    public virtual ValueTask DisposeAsync() => ValueTask.CompletedTask;
 193
 194    /// <summary>Runs the ExecuteHandlerAsync operation.</summary>
 195    protected async Task ExecuteHandlerAsync(
 196        RedisStreamDelivery delivery,
 197        CancellationToken cancellationToken,
 198        bool logFailures = true)
 199    {
 200        using var activity = AsyncResponseDiagnostics.StartActivity(
 201            "asyncresponse.redis.receive",
 202            ActivityKind.Consumer,
 203            delivery.CorrelationId);
 204        activity?.SetTag("asyncresponse.transport", "redis");
 205        activity?.SetTag("asyncresponse.redis.role", _role.ToString());
 206        activity?.SetTag("asyncresponse.redis.ack_mode", _subscriberOptions.AckMode.ToString());
 207        activity?.SetTag("asyncresponse.redis.delivery_attempt", delivery.Attempt);
 208        activity?.SetTag("messaging.system", "redis");
 209        activity?.SetTag("messaging.destination.name", _stream);
 210        activity?.SetTag("messaging.message.id", delivery.MessageId.ToString());
 211
 212        try
 213        {
 214            await _handler(delivery, cancellationToken).ConfigureAwait(false);
 215        }
 216        catch (Exception ex)
 217        {
 218            if (logFailures)
 219            {
 220                Logger.LogError(
 221                    ex,
 222                    "Redis stream message handling failed for {Stream}/{MessageId}.",
 223                    _stream,
 224                    delivery.MessageId.ToString());
 225            }
 226
 227            AsyncResponseDiagnostics.SetError(activity, ex);
 228            throw;
 229        }
 230    }
 231
 232    /// <summary>Acknowledges the delivered message.</summary>
 233    protected Task AckAsync(RedisStreamDelivery delivery, CancellationToken cancellationToken)
 234        => _database.StreamAcknowledgeAsync(
 235            delivery.Stream,
 236            delivery.ConsumerGroup,
 237            delivery.MessageId,
 238            cancellationToken);
 239
 240    /// <summary>Runs the AlreadyExceededDeliveryAttempts operation.</summary>
 241    protected bool AlreadyExceededDeliveryAttempts(RedisStreamDelivery delivery)
 242        => MaxDeliveryAttempts > 0 && delivery.Attempt > MaxDeliveryAttempts;
 243
 244    /// <summary>Runs the ReachedDeliveryAttempts operation.</summary>
 245    protected bool ReachedDeliveryAttempts(RedisStreamDelivery delivery)
 246        => MaxDeliveryAttempts > 0 && delivery.Attempt >= MaxDeliveryAttempts;
 247
 248    /// <summary>Moves the delivered message to dead-letter storage and acknowledges it.</summary>
 249    protected async Task DeadLetterAndAckAsync(
 250        RedisStreamDelivery delivery,
 251        Exception exception,
 252        string reason,
 253        CancellationToken cancellationToken)
 254    {
 255        if (TransportOptions.DeadLetterEnabled)
 256        {
 257            var fields = new[]
 258            {
 259                new NameValueEntry("sourceStream", delivery.Stream.ToString()),
 260                new NameValueEntry("consumerGroup", delivery.ConsumerGroup.ToString()),
 261                new NameValueEntry("subscriberRole", _role.ToString()),
 262                new NameValueEntry("messageId", delivery.MessageId.ToString()),
 263                new NameValueEntry("correlationId", delivery.CorrelationId ?? string.Empty),
 264                new NameValueEntry("attempt", delivery.Attempt),
 265                new NameValueEntry("reason", reason),
 266                new NameValueEntry("exceptionType", exception.GetType().FullName!),
 267                new NameValueEntry("exceptionMessage", exception.Message),
 268                new NameValueEntry("payload", delivery.Payload),
 269                new NameValueEntry("occurredAtUtc", DateTimeOffset.UtcNow.ToString("O"))
 270            };
 271
 272            await _database.StreamAddAsync(
 273                _keys.DeadLetterStream,
 274                fields,
 275                TransportOptions.DeadLetterStreamMaxLength,
 276                TransportOptions.UseApproximateStreamTrimming,
 277                cancellationToken).ConfigureAwait(false);
 278        }
 279
 280        await AckAsync(delivery, cancellationToken).ConfigureAwait(false);
 281    }
 282
 283    /// <summary>
 284    /// Dead-letters (when enabled) and ACKs a stream entry that could not be turned into a delivery —
 285    /// for example a foreign or malformed entry with no payload field, or a tombstone left behind when
 286    /// trimming evicts a still-pending entry. Without this, such an entry throws before
 287    /// <see cref="HandleAsync"/> runs, so it is never ACKed: the pending-claim loop re-claims it every
 288    /// cycle and the subscriber faults and restarts indefinitely while the entry never drains.
 289    /// </summary>
 290    public async Task DiscardUnprocessableAsync(
 291        RedisKey stream,
 292        RedisValue consumerGroup,
 293        StreamEntry entry,
 294        Exception failure,
 295        CancellationToken cancellationToken)
 296    {
 297        if (entry.Id.IsNull)
 298        {
 299            // A trimmed-while-pending tombstone (Redis 5/6 answer XCLAIM with a nil entry) carries
 300            // no id to ACK and no payload to record: sending its null id to XACK is rejected by the
 301            // client from inside the caller's catch, which replaced the original error, faulted the
 302            // subscriber, and re-dead-lettered the tombstone every claim cycle. The claim loop
 303            // drains it by its pending id instead; nothing to settle here.
 304            Logger.LogDebug(failure, "Redis claim on {Stream} returned a trimmed tombstone; skipping it.", _stream);
 305            return;
 306        }
 307
 308        Logger.LogError(
 309            failure,
 310            "Redis entry {MessageId} on {Stream} could not be parsed into a delivery; dead-lettering and ACKing it to av
 311            entry.Id.ToString(),
 312            _stream);
 313
 314        var delivery = new RedisStreamDelivery(
 315            stream,
 316            consumerGroup,
 317            entry.Id,
 318            DescribeRawEntry(entry),
 319            RedisCorrelationIdExtractor.TryReadField(entry, TransportOptions.CorrelationIdField),
 320            Attempt: 0,
 321            entry);
 322
 323        // Settlement deliberately ignores cancellation (as every other settlement in this file
 324        // does): a shutdown landing between the dead-letter XADD and the XACK left the entry in
 325        // the PEL to be reclaimed and dead-lettered a SECOND time after restart.
 326        await DeadLetterAndAckAsync(delivery, failure, "unparsable_entry", CancellationToken.None).ConfigureAwait(false)
 327    }
 328
 329    private static string DescribeRawEntry(StreamEntry entry)
 330        => entry.Values is { Length: > 0 }
 331            ? string.Join("; ", entry.Values.Select(value => $"{value.Name}={value.Value}"))
 332            : string.Empty;
 333
 334    /// <summary>Runs the NotifyBackgroundFailureAsync operation.</summary>
 335    protected async ValueTask NotifyBackgroundFailureAsync(
 336        RedisStreamDelivery delivery,
 337        Exception exception)
 338    {
 339        var callback = _subscriberOptions.OnBackgroundFailure;
 340        if (callback is null)
 341            return;
 342
 343        try
 344        {
 345            await callback(new RedisBackgroundFailureContext(
 346                _stream,
 347                _consumerGroup,
 348                _role.ToString(),
 349                delivery.MessageId.ToString(),
 350                delivery.CorrelationId,
 351                exception)).ConfigureAwait(false);
 352        }
 353        catch (Exception callbackException)
 354        {
 355            Logger.LogError(
 356                callbackException,
 357                "Redis background failure callback failed for already-ACKed message {MessageId} on {Stream}.",
 358                delivery.MessageId.ToString(),
 359                _stream);
 360        }
 361    }
 362}
 363
 364internal sealed class AwaitingRedisMessageDispatcher(
 365    Func<RedisStreamDelivery, CancellationToken, Task> handler,
 366    IRedisStreamDatabase database,
 367    RedisAsyncResponseTransportOptions transportOptions,
 368    RedisSubscriberOptions subscriberOptions,
 369    ILogger logger,
 370    RedisKey stream,
 371    RedisValue consumerGroup,
 372    RedisSubscriberRole role)
 446373    : RedisMessageDispatcher(handler, database, transportOptions, subscriberOptions, logger, stream, consumerGroup, role
 374{
 375    /// <summary>Handles the delivered message.</summary>
 376    public override async Task<RedisDispatchOutcome> HandleAsync(
 377        RedisStreamDelivery delivery,
 378        CancellationToken subscriberCancellationToken)
 379    {
 458380        if (AlreadyExceededDeliveryAttempts(delivery))
 381        {
 8382            await TryDeadLetterAndAckAsync(
 8383                delivery,
 8384                new InvalidOperationException($"Redis message exceeded {MaxDeliveryAttempts} delivery attempts."),
 8385                "max_delivery_attempts_exceeded").ConfigureAwait(false);
 8386            return RedisDispatchOutcome.Processed;
 387        }
 388
 389        try
 390        {
 450391            await ExecuteHandlerAsync(delivery, subscriberCancellationToken).ConfigureAwait(false);
 432392        }
 2393        catch (OperationCanceledException) when (subscriberCancellationToken.IsCancellationRequested)
 394        {
 2395            throw;
 396        }
 16397        catch (Exception ex) when (ReachedDeliveryAttempts(delivery))
 398        {
 9399            Logger.LogWarning(
 9400                ex,
 9401                "Redis message {MessageId} reached max delivery attempts ({MaxDeliveryAttempts}); writing to dead-letter
 9402                delivery.MessageId.ToString(),
 9403                MaxDeliveryAttempts);
 9404            await TryDeadLetterAndAckAsync(delivery, ex, "handler_failed_max_attempts").ConfigureAwait(false);
 9405            return RedisDispatchOutcome.Processed;
 406        }
 7407        catch
 408        {
 409            // Leave the entry pending. The subscriber's pending-claim loop reclaims it after
 410            // PendingMessageMinIdleTime, giving Redis-backed retry without a hot loop.
 7411            return RedisDispatchOutcome.Processed;
 412        }
 413
 414        // The ACK sits outside the handler's try/catch: a transient XACK failure after a
 415        // successful handler must not be misread as a handler failure — dead-lettering or leaving
 416        // it for reclaim here would redeliver (or bury) work whose side effects already completed.
 417        // Swallow and log instead; the entry stays pending and at-least-once redelivery applies.
 418        // Settlement deliberately ignores cancellation (as every sibling transport does): the
 419        // handler already completed, and abandoning the XACK on shutdown leaves the entry in the
 420        // PEL to be reclaimed and re-run after restart.
 421        try
 422        {
 432423            await AckAsync(delivery, CancellationToken.None).ConfigureAwait(false);
 430424        }
 2425        catch (Exception ex)
 426        {
 2427            Logger.LogWarning(
 2428                ex,
 2429                "Failed to ACK Redis message {MessageId} on {Stream} after a successful handler; the entry stays pending
 2430                delivery.MessageId.ToString(),
 2431                delivery.Stream.ToString());
 2432        }
 433
 432434        return RedisDispatchOutcome.Processed;
 456435    }
 436
 437    /// <summary>
 438    /// Burial that never throws (queued-dispatcher and DB-transport parity: "a burial that throws
 439    /// is a burial that failed"). Unguarded, a dead-letter XADD that failed — MISCONF/OOM, the
 440    /// adapter's timeout, a WRONGTYPE on the dead-letter key — escaped past the XACK to the
 441    /// supervisor, which restarted the subscriber; the pending-claim loop then re-claimed the
 442    /// same entry every cycle and the whole stream stopped draining. Settlement deliberately
 443    /// ignores cancellation: a shutdown landing between the XADD and the XACK would leave the
 444    /// entry in the PEL to be reclaimed and dead-lettered a SECOND time.
 445    /// </summary>
 446    private async Task TryDeadLetterAndAckAsync(RedisStreamDelivery delivery, Exception exception, string reason)
 447    {
 448        try
 449        {
 17450            await DeadLetterAndAckAsync(delivery, exception, reason, CancellationToken.None).ConfigureAwait(false);
 13451        }
 4452        catch (Exception deadLetterException)
 453        {
 4454            Logger.LogError(
 4455                deadLetterException,
 4456                "Failed to dead-letter Redis message {MessageId} on {Stream} ({Reason}); the entry stays pending and is 
 4457                delivery.MessageId.ToString(),
 4458                delivery.Stream.ToString(),
 4459                reason);
 4460        }
 17461    }
 462}
 463
 464internal sealed class QueuedRedisMessageDispatcher : RedisMessageDispatcher
 465{
 466    private readonly Channel<RedisStreamDelivery> _queue;
 467    private readonly Task[] _workers;
 468    private readonly CancellationTokenSource _drainCancellation = new();
 469    private readonly TimeSpan _drainTimeout;
 470    private readonly int _capacity;
 471    private readonly string _stream;
 472    private int _pendingCount;
 473    private int _runningCount;
 474    private int _disposeStarted;
 475
 476    /// <summary>Runs the QueuedRedisMessageDispatcher operation.</summary>
 477    public QueuedRedisMessageDispatcher(
 478        Func<RedisStreamDelivery, CancellationToken, Task> handler,
 479        IRedisStreamDatabase database,
 480        RedisAsyncResponseTransportOptions transportOptions,
 481        RedisSubscriberOptions subscriberOptions,
 482        ILogger logger,
 483        RedisKey stream,
 484        RedisValue consumerGroup,
 485        RedisSubscriberRole role)
 486        : base(handler, database, transportOptions, subscriberOptions, logger, stream, consumerGroup, role)
 487    {
 488        _stream = stream.ToString();
 489        _drainTimeout = subscriberOptions.BackgroundDrainTimeout;
 490        _capacity = subscriberOptions.BackgroundQueueCapacity;
 491        _queue = Channel.CreateBounded<RedisStreamDelivery>(new BoundedChannelOptions(subscriberOptions.BackgroundQueueC
 492        {
 493            AllowSynchronousContinuations = false,
 494            FullMode = BoundedChannelFullMode.Wait,
 495            SingleReader = subscriberOptions.BackgroundWorkerCount == 1,
 496            SingleWriter = false
 497        });
 498
 499        _workers = Enumerable.Range(0, subscriberOptions.BackgroundWorkerCount)
 500            .Select(workerIndex => Task.Run(() => RunWorkerAsync(workerIndex)))
 501            .ToArray();
 502
 503        Logger.LogInformation(
 504            "Created Redis ACK-after-enqueue dispatcher for {Stream} with {WorkerCount} worker(s), queue capacity {Queue
 505            _stream,
 506            subscriberOptions.BackgroundWorkerCount,
 507            subscriberOptions.BackgroundQueueCapacity,
 508            _drainTimeout);
 509    }
 510
 511    internal int PendingCount => Volatile.Read(ref _pendingCount);
 512    internal int RunningCount => Volatile.Read(ref _runningCount);
 513
 514    public override bool CanAcceptMore => Volatile.Read(ref _pendingCount) < _capacity;
 515
 516    public override int FreeCapacity => Math.Max(0, _capacity - Volatile.Read(ref _pendingCount));
 517
 518    /// <summary>Handles the delivered message.</summary>
 519    public override async Task<RedisDispatchOutcome> HandleAsync(
 520        RedisStreamDelivery delivery,
 521        CancellationToken subscriberCancellationToken)
 522    {
 523        // Pre-execution cap, BEFORE the enqueue-and-ACK (awaiting-dispatcher parity): the
 524        // pending-claim loop feeds this dispatcher real XPENDING delivery counts too, and without
 525        // the check an over-cap entry — deferred under backpressure and reclaimed each cycle, or
 526        // re-claimed after a swallowed post-enqueue ACK failure — was re-enqueued and re-executed
 527        // forever, with nothing ever consulting MaxDeliveryAttempts to bury it.
 528        if (AlreadyExceededDeliveryAttempts(delivery))
 529        {
 530            await DeadLetterAndAckAsync(
 531                delivery,
 532                new InvalidOperationException($"Redis message exceeded {MaxDeliveryAttempts} delivery attempts."),
 533                "max_delivery_attempts_exceeded",
 534                CancellationToken.None).ConfigureAwait(false);
 535            return RedisDispatchOutcome.Processed;
 536        }
 537
 538        Interlocked.Increment(ref _pendingCount);
 539        if (!_queue.Writer.TryWrite(delivery))
 540        {
 541            Interlocked.Decrement(ref _pendingCount);
 542            Logger.LogWarning(
 543                "Redis background queue rejected message {MessageId} for {Stream}; leaving it pending for retry. Pending
 544                delivery.MessageId.ToString(),
 545                _stream,
 546                PendingCount,
 547                RunningCount);
 548            return RedisDispatchOutcome.Deferred;
 549        }
 550
 551        // The entry now belongs to a background worker, which decrements _pendingCount when it dequeues.
 552        // Do not touch the counter again here, even if the ACK below fails — otherwise it double-counts.
 553        // Settlement deliberately ignores cancellation (as every sibling transport does): a graceful
 554        // shutdown drains the background queue and runs this entry, so abandoning the XACK on the
 555        // stopping token would leave completed work in the PEL to be reclaimed and re-run.
 556        try
 557        {
 558            await AckAsync(delivery, CancellationToken.None).ConfigureAwait(false);
 559        }
 560        catch (Exception ex)
 561        {
 562            Logger.LogError(
 563                ex,
 564                "Failed to ACK Redis message {MessageId} for {Stream} after enqueue; it is being processed but Redis wil
 565                delivery.MessageId.ToString(),
 566                _stream);
 567        }
 568
 569        return RedisDispatchOutcome.Processed;
 570    }
 571
 572    /// <summary>Releases resources held by this instance.</summary>
 573    public override async ValueTask DisposeAsync()
 574    {
 575        if (Interlocked.Exchange(ref _disposeStarted, 1) != 0)
 576            return;
 577
 578        Logger.LogInformation(
 579            "Draining Redis ACK-after-enqueue dispatcher for {Stream}. Pending={PendingCount}, Running={RunningCount}.",
 580            _stream,
 581            PendingCount,
 582            RunningCount);
 583        _queue.Writer.TryComplete();
 584
 585        try
 586        {
 587            await Task.WhenAll(_workers).WaitAsync(_drainTimeout).ConfigureAwait(false);
 588            _drainCancellation.Dispose();
 589        }
 590        catch (TimeoutException ex)
 591        {
 592            _drainCancellation.Cancel();
 593            Logger.LogWarning(
 594                ex,
 595                "Timed out while draining Redis ACK-after-enqueue dispatcher for {Stream}. Pending={PendingCount}, Runni
 596                _stream,
 597                PendingCount,
 598                RunningCount);
 599
 600            _ = Task.WhenAll(_workers).ContinueWith(
 601                _ => _drainCancellation.Dispose(),
 602                CancellationToken.None,
 603                TaskContinuationOptions.ExecuteSynchronously,
 604                TaskScheduler.Default);
 605        }
 606        catch (Exception ex)
 607        {
 608            // A worker faulted outside its own handler guard (DB/NATS dispatcher parity). WhenAll
 609            // only completes once every worker has finished, so the source is safe to dispose here
 610            // — and the fault must not escape DisposeAsync and mask the real shutdown path.
 611            Logger.LogDebug(ex, "Redis ACK-after-enqueue dispatcher drain for {Stream} ended with an error.", _stream);
 612            _drainCancellation.Dispose();
 613        }
 614    }
 615
 616    private async Task RunWorkerAsync(int workerIndex)
 617    {
 618        await foreach (var delivery in _queue.Reader.ReadAllAsync().ConfigureAwait(false))
 619        {
 620            Interlocked.Decrement(ref _pendingCount);
 621            Interlocked.Increment(ref _runningCount);
 622
 623            // Once the drain budget has lapsed, STOP executing. The token below cannot stop the
 624            // real handler — it is `_ingress.HandleWorkerMessageAsync(payload)`, whose target takes
 625            // no CancellationToken — so nothing ever raised the OperationCanceledException the arm
 626            // below was written for, the loop kept starting fresh work past the budget, and every
 627            // entry still queued at process exit vanished with no record (they were ACKed at
 628            // enqueue, so Redis will not redeliver them). Route them through the same
 629            // OnBackgroundFailure/dead-letter path instead of losing them silently.
 630            if (_drainCancellation.IsCancellationRequested)
 631            {
 632                var lapsed = new OperationCanceledException(
 633                    "The ACK-after-enqueue drain budget lapsed before this already-ACKed message was handled.");
 634
 635                Logger.LogWarning(
 636                    "Redis background handler for already-ACKed message {MessageId} on {Stream} was not started: the dis
 637                    delivery.MessageId.ToString(),
 638                    _stream);
 639
 640                await NotifyBackgroundFailureAsync(delivery, lapsed).ConfigureAwait(false);
 641
 642                try
 643                {
 644                    await DeadLetterAndAckAsync(
 645                        delivery,
 646                        lapsed,
 647                        "drain_budget_lapsed_after_ack",
 648                        CancellationToken.None).ConfigureAwait(false);
 649                }
 650                catch (Exception deadLetterException)
 651                {
 652                    Logger.LogError(
 653                        deadLetterException,
 654                        "Failed to dead-letter undrained Redis message {MessageId} on {Stream}.",
 655                        delivery.MessageId.ToString(),
 656                        _stream);
 657                }
 658
 659                Interlocked.Decrement(ref _runningCount);
 660                continue;
 661            }
 662
 663            try
 664            {
 665                await ExecuteHandlerAsync(
 666                    delivery,
 667                    _drainCancellation.Token,
 668                    logFailures: false).ConfigureAwait(false);
 669            }
 670            catch (OperationCanceledException ex) when (_drainCancellation.IsCancellationRequested)
 671            {
 672                // The drain budget lapsed with this already-ACKed entry still unprocessed: Redis
 673                // will not redeliver it, so surface the drop through OnBackgroundFailure instead of
 674                // losing it silently.
 675                Logger.LogWarning(
 676                    "Redis background handler for already-ACKed message {MessageId} on {Stream} was canceled during disp
 677                    delivery.MessageId.ToString(),
 678                    _stream);
 679                await NotifyBackgroundFailureAsync(delivery, ex).ConfigureAwait(false);
 680            }
 681            catch (Exception ex)
 682            {
 683                Logger.LogError(
 684                    ex,
 685                    "Redis background handler failed for already-ACKed message {MessageId} on {Stream}.",
 686                    delivery.MessageId.ToString(),
 687                    _stream);
 688                await NotifyBackgroundFailureAsync(delivery, ex).ConfigureAwait(false);
 689
 690                try
 691                {
 692                    await DeadLetterAndAckAsync(
 693                        delivery,
 694                        ex,
 695                        "background_handler_failed_after_ack",
 696                        CancellationToken.None).ConfigureAwait(false);
 697                }
 698                catch (Exception deadLetterException)
 699                {
 700                    Logger.LogError(
 701                        deadLetterException,
 702                        "Failed to dead-letter already-ACKed Redis message {MessageId} on {Stream}.",
 703                        delivery.MessageId.ToString(),
 704                        _stream);
 705                }
 706            }
 707            finally
 708            {
 709                Interlocked.Decrement(ref _runningCount);
 710            }
 711        }
 712    }
 713}
Generated by: ReportGenerator 5.5.11.0
09/22/2026 - 09:41:04
GitHub | reportgenerator.io

Methods/Properties

.ctor(System.Func`3<AsyncResponse.Transports.Redis.RedisStreamDelivery,System.Threading.CancellationToken,System.Threading.Tasks.Task>,AsyncResponse.Transports.Redis.IRedisStreamDatabase,AsyncResponse.Transports.Redis.RedisAsyncResponseTransportOptions,AsyncResponse.Transports.Redis.RedisSubscriberOptions,Microsoft.Extensions.Logging.ILogger,StackExchange.Redis.RedisKey,StackExchange.Redis.RedisValue,AsyncResponse.Transports.Redis.RedisSubscriberRole)
HandleAsync()
TryDeadLetterAndAckAsync()