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

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

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%11100%
get_TransportOptions()100%11100%
get_Logger()100%11100%
get_MaxDeliveryAttempts()100%11100%
Create(...)100%22100%
ValidateOptions(...)100%1616100%
get_CanAcceptMore()100%11100%
get_FreeCapacity()100%11100%
DisposeAsync()100%11100%
ExecuteHandlerAsync()100%1616100%
AckAsync(...)100%11100%
AlreadyExceededDeliveryAttempts(...)100%22100%
ReachedDeliveryAttempts(...)100%22100%
DeadLetterAndAckAsync()100%44100%
DiscardUnprocessableAsync()100%22100%
DescribeRawEntry(...)50%44100%
NotifyBackgroundFailureAsync()100%22100%

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>
 48043    protected RedisMessageDispatcher(
 48044        Func<RedisStreamDelivery, CancellationToken, Task> handler,
 48045        IRedisStreamDatabase database,
 48046        RedisAsyncResponseTransportOptions transportOptions,
 48047        RedisSubscriberOptions subscriberOptions,
 48048        ILogger logger,
 48049        RedisKey stream,
 48050        RedisValue consumerGroup,
 48051        RedisSubscriberRole role)
 52    {
 48053        _handler = handler;
 48054        _database = database;
 48055        TransportOptions = transportOptions;
 48056        _subscriberOptions = subscriberOptions;
 48057        _keys = new RedisTransportKeySchema(transportOptions);
 48058        Logger = logger;
 48059        _stream = stream.ToString();
 48060        _consumerGroup = consumerGroup.ToString();
 48061        _role = role;
 48062    }
 63
 10764    protected RedisAsyncResponseTransportOptions TransportOptions { get; }
 14365    protected ILogger Logger { get; }
 66
 107567    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    {
 47880        ValidateOptions(transportOptions, subscriberOptions, role);
 81
 47882        if (subscriberOptions.AckMode is RedisAckMode.AckAfterEnqueue)
 83        {
 3484            return new QueuedRedisMessageDispatcher(
 3485                handler,
 3486                database,
 3487                transportOptions,
 3488                subscriberOptions,
 3489                logger,
 3490                stream,
 3491                consumerGroup,
 3492                role);
 93        }
 94
 44495        return new AwaitingRedisMessageDispatcher(
 44496            handler,
 44497            database,
 44498            transportOptions,
 44499            subscriberOptions,
 444100            logger,
 444101            stream,
 444102            consumerGroup,
 444103            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    {
 924112        RedisTransportOptionsValidator.ValidateCommon(transportOptions);
 113
 924114        var optionPath = role is RedisSubscriberRole.Worker
 924115            ? $"{nameof(RedisAsyncResponseTransportOptions)}.{nameof(RedisAsyncResponseTransportOptions.WorkerSubscriber
 924116            : $"{nameof(RedisAsyncResponseTransportOptions)}.{nameof(RedisAsyncResponseTransportOptions.ResponseSubscrib
 117
 924118        if (subscriberOptions.BatchSize <= 0)
 2119            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.
 922126        AsyncResponseChannelOptions.EnsureTimerBacked(subscriberOptions.EmptyPollDelay, optionPath, nameof(RedisSubscrib
 918127        AsyncResponseChannelOptions.EnsureTimerBacked(subscriberOptions.PendingMessageMinIdleTime, optionPath, nameof(Re
 914128        AsyncResponseChannelOptions.EnsurePersistedTtl(subscriberOptions.PendingClaimInterval, optionPath, nameof(RedisS
 912129        if (subscriberOptions.PendingClaimBatchSize <= 0)
 2130            throw new InvalidOperationException($"{optionPath}.{nameof(RedisSubscriberOptions.PendingClaimBatchSize)} mu
 910131        if (subscriberOptions.MaxDeliveryAttempts < 0)
 2132            throw new InvalidOperationException($"{optionPath}.{nameof(RedisSubscriberOptions.MaxDeliveryAttempts)} cann
 133
 908134        switch (subscriberOptions.AckMode)
 135        {
 136            case RedisAckMode.AckAfterHandlerCompletes:
 852137                return;
 138
 139            case RedisAckMode.AckAfterEnqueue:
 54140                if (subscriberOptions.BackgroundWorkerCount <= 0)
 141                {
 4142                    throw new InvalidOperationException(
 4143                        $"{optionPath}.{nameof(RedisSubscriberOptions.BackgroundWorkerCount)} must be explicitly configu
 4144                        $"when {nameof(RedisSubscriberOptions.AckMode)} is {nameof(RedisAckMode.AckAfterEnqueue)}.");
 145                }
 146
 50147                if (subscriberOptions.BackgroundQueueCapacity <= 0)
 148                {
 2149                    throw new InvalidOperationException(
 2150                        $"{optionPath}.{nameof(RedisSubscriberOptions.BackgroundQueueCapacity)} must be explicitly confi
 2151                        $"when {nameof(RedisSubscriberOptions.AckMode)} is {nameof(RedisAckMode.AckAfterEnqueue)}.");
 152                }
 153
 48154                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.
 46158                ShutdownBudgetValidator.Validate(
 46159                    "Redis",
 46160                    $"{nameof(RedisAsyncResponseTransportOptions)}.{nameof(RedisAsyncResponseTransportOptions.HostShutdo
 46161                    transportOptions.HostShutdownTimeout,
 46162                    ($"{optionPath}.{nameof(RedisSubscriberOptions.BackgroundDrainTimeout)}", subscriberOptions.Backgrou
 163
 44164                return;
 165
 166            default:
 2167                throw new InvalidOperationException(
 2168                    $"{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>
 4166183    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>
 4584189    public virtual int FreeCapacity => int.MaxValue;
 190
 191    /// <summary>Releases resources held by this instance.</summary>
 444192    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    {
 496200        using var activity = AsyncResponseDiagnostics.StartActivity(
 496201            "asyncresponse.redis.receive",
 496202            ActivityKind.Consumer,
 496203            delivery.CorrelationId);
 496204        activity?.SetTag("asyncresponse.transport", "redis");
 496205        activity?.SetTag("asyncresponse.redis.role", _role.ToString());
 496206        activity?.SetTag("asyncresponse.redis.ack_mode", _subscriberOptions.AckMode.ToString());
 496207        activity?.SetTag("asyncresponse.redis.delivery_attempt", delivery.Attempt);
 496208        activity?.SetTag("messaging.system", "redis");
 496209        activity?.SetTag("messaging.destination.name", _stream);
 496210        activity?.SetTag("messaging.message.id", delivery.MessageId.ToString());
 211
 212        try
 213        {
 496214            await _handler(delivery, cancellationToken).ConfigureAwait(false);
 466215        }
 30216        catch (Exception ex)
 217        {
 30218            if (logFailures)
 219            {
 18220                Logger.LogError(
 18221                    ex,
 18222                    "Redis stream message handling failed for {Stream}/{MessageId}.",
 18223                    _stream,
 18224                    delivery.MessageId.ToString());
 225            }
 226
 30227            AsyncResponseDiagnostics.SetError(activity, ex);
 30228            throw;
 229        }
 466230    }
 231
 232    /// <summary>Acknowledges the delivered message.</summary>
 233    protected Task AckAsync(RedisStreamDelivery delivery, CancellationToken cancellationToken)
 511234        => _database.StreamAcknowledgeAsync(
 511235            delivery.Stream,
 511236            delivery.ConsumerGroup,
 511237            delivery.MessageId,
 511238            cancellationToken);
 239
 240    /// <summary>Runs the AlreadyExceededDeliveryAttempts operation.</summary>
 241    protected bool AlreadyExceededDeliveryAttempts(RedisStreamDelivery delivery)
 514242        => MaxDeliveryAttempts > 0 && delivery.Attempt > MaxDeliveryAttempts;
 243
 244    /// <summary>Runs the ReachedDeliveryAttempts operation.</summary>
 245    protected bool ReachedDeliveryAttempts(RedisStreamDelivery delivery)
 16246        => 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    {
 35255        if (TransportOptions.DeadLetterEnabled)
 256        {
 33257            var fields = new[]
 33258            {
 33259                new NameValueEntry("sourceStream", delivery.Stream.ToString()),
 33260                new NameValueEntry("consumerGroup", delivery.ConsumerGroup.ToString()),
 33261                new NameValueEntry("subscriberRole", _role.ToString()),
 33262                new NameValueEntry("messageId", delivery.MessageId.ToString()),
 33263                new NameValueEntry("correlationId", delivery.CorrelationId ?? string.Empty),
 33264                new NameValueEntry("attempt", delivery.Attempt),
 33265                new NameValueEntry("reason", reason),
 33266                new NameValueEntry("exceptionType", exception.GetType().FullName!),
 33267                new NameValueEntry("exceptionMessage", exception.Message),
 33268                new NameValueEntry("payload", delivery.Payload),
 33269                new NameValueEntry("occurredAtUtc", DateTimeOffset.UtcNow.ToString("O"))
 33270            };
 271
 33272            await _database.StreamAddAsync(
 33273                _keys.DeadLetterStream,
 33274                fields,
 33275                TransportOptions.DeadLetterStreamMaxLength,
 33276                TransportOptions.UseApproximateStreamTrimming,
 33277                cancellationToken).ConfigureAwait(false);
 278        }
 279
 29280        await AckAsync(delivery, cancellationToken).ConfigureAwait(false);
 29281    }
 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    {
 8297        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.
 2304            Logger.LogDebug(failure, "Redis claim on {Stream} returned a trimmed tombstone; skipping it.", _stream);
 2305            return;
 306        }
 307
 6308        Logger.LogError(
 6309            failure,
 6310            "Redis entry {MessageId} on {Stream} could not be parsed into a delivery; dead-lettering and ACKing it to av
 6311            entry.Id.ToString(),
 6312            _stream);
 313
 6314        var delivery = new RedisStreamDelivery(
 6315            stream,
 6316            consumerGroup,
 6317            entry.Id,
 6318            DescribeRawEntry(entry),
 6319            RedisCorrelationIdExtractor.TryReadField(entry, TransportOptions.CorrelationIdField),
 6320            Attempt: 0,
 6321            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.
 6326        await DeadLetterAndAckAsync(delivery, failure, "unparsable_entry", CancellationToken.None).ConfigureAwait(false)
 8327    }
 328
 329    private static string DescribeRawEntry(StreamEntry entry)
 6330        => entry.Values is { Length: > 0 }
 6331            ? string.Join("; ", entry.Values.Select(value => $"{value.Name}={value.Value}"))
 6332            : string.Empty;
 333
 334    /// <summary>Runs the NotifyBackgroundFailureAsync operation.</summary>
 335    protected async ValueTask NotifyBackgroundFailureAsync(
 336        RedisStreamDelivery delivery,
 337        Exception exception)
 338    {
 16339        var callback = _subscriberOptions.OnBackgroundFailure;
 16340        if (callback is null)
 4341            return;
 342
 343        try
 344        {
 12345            await callback(new RedisBackgroundFailureContext(
 12346                _stream,
 12347                _consumerGroup,
 12348                _role.ToString(),
 12349                delivery.MessageId.ToString(),
 12350                delivery.CorrelationId,
 12351                exception)).ConfigureAwait(false);
 10352        }
 2353        catch (Exception callbackException)
 354        {
 2355            Logger.LogError(
 2356                callbackException,
 2357                "Redis background failure callback failed for already-ACKed message {MessageId} on {Stream}.",
 2358                delivery.MessageId.ToString(),
 2359                _stream);
 2360        }
 16361    }
 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)
 373    : 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    {
 380        if (AlreadyExceededDeliveryAttempts(delivery))
 381        {
 382            await TryDeadLetterAndAckAsync(
 383                delivery,
 384                new InvalidOperationException($"Redis message exceeded {MaxDeliveryAttempts} delivery attempts."),
 385                "max_delivery_attempts_exceeded").ConfigureAwait(false);
 386            return RedisDispatchOutcome.Processed;
 387        }
 388
 389        try
 390        {
 391            await ExecuteHandlerAsync(delivery, subscriberCancellationToken).ConfigureAwait(false);
 392        }
 393        catch (OperationCanceledException) when (subscriberCancellationToken.IsCancellationRequested)
 394        {
 395            throw;
 396        }
 397        catch (Exception ex) when (ReachedDeliveryAttempts(delivery))
 398        {
 399            Logger.LogWarning(
 400                ex,
 401                "Redis message {MessageId} reached max delivery attempts ({MaxDeliveryAttempts}); writing to dead-letter
 402                delivery.MessageId.ToString(),
 403                MaxDeliveryAttempts);
 404            await TryDeadLetterAndAckAsync(delivery, ex, "handler_failed_max_attempts").ConfigureAwait(false);
 405            return RedisDispatchOutcome.Processed;
 406        }
 407        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.
 411            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        {
 423            await AckAsync(delivery, CancellationToken.None).ConfigureAwait(false);
 424        }
 425        catch (Exception ex)
 426        {
 427            Logger.LogWarning(
 428                ex,
 429                "Failed to ACK Redis message {MessageId} on {Stream} after a successful handler; the entry stays pending
 430                delivery.MessageId.ToString(),
 431                delivery.Stream.ToString());
 432        }
 433
 434        return RedisDispatchOutcome.Processed;
 435    }
 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        {
 450            await DeadLetterAndAckAsync(delivery, exception, reason, CancellationToken.None).ConfigureAwait(false);
 451        }
 452        catch (Exception deadLetterException)
 453        {
 454            Logger.LogError(
 455                deadLetterException,
 456                "Failed to dead-letter Redis message {MessageId} on {Stream} ({Reason}); the entry stays pending and is 
 457                delivery.MessageId.ToString(),
 458                delivery.Stream.ToString(),
 459                reason);
 460        }
 461    }
 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)
get_TransportOptions()
get_Logger()
get_MaxDeliveryAttempts()
Create(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)
ValidateOptions(AsyncResponse.Transports.Redis.RedisAsyncResponseTransportOptions,AsyncResponse.Transports.Redis.RedisSubscriberOptions,AsyncResponse.Transports.Redis.RedisSubscriberRole)
get_CanAcceptMore()
get_FreeCapacity()
DisposeAsync()
ExecuteHandlerAsync()
AckAsync(AsyncResponse.Transports.Redis.RedisStreamDelivery,System.Threading.CancellationToken)
AlreadyExceededDeliveryAttempts(AsyncResponse.Transports.Redis.RedisStreamDelivery)
ReachedDeliveryAttempts(AsyncResponse.Transports.Redis.RedisStreamDelivery)
DeadLetterAndAckAsync()
DiscardUnprocessableAsync()
DescribeRawEntry(StackExchange.Redis.StreamEntry)
NotifyBackgroundFailureAsync()