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

Information
Class: AsyncResponse.Transports.SQS.SqsSubscriberService.BatchProgress
Assembly: AsyncResponse.Transports.SQS
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.SQS/SqsSubscriberServices.cs
Line coverage
100%
Covered lines: 5
Uncovered lines: 0
Coverable lines: 5
Total lines: 300
Line coverage: 100%
Branch coverage
N/A
Covered branches: 0
Total branches: 0
Branch coverage: N/A
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%11100%
get_SettledCount()100%11100%
MarkSettled()100%11100%
SuppressRenewal(...)100%11100%
IsRenewalSuppressed(...)100%11100%

File(s)

/home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.SQS/SqsSubscriberServices.cs

#LineLine coverage
 1using Microsoft.Extensions.Hosting;
 2using Microsoft.Extensions.Logging;
 3using Microsoft.Extensions.Options;
 4
 5namespace AsyncResponse.Transports.SQS;
 6
 7internal abstract class SqsSubscriberService : BackgroundService
 8{
 9    private readonly ISqsClient _client;
 10
 11    protected SqsSubscriberService(
 12        IOptions<SqsAsyncResponseOptions> options,
 13        ISqsClient client,
 14        ILogger logger)
 15    {
 16        Options = options.Value;
 17        SqsOptionsValidator.ValidateCommon(Options);
 18        _client = client;
 19        Logger = logger;
 20    }
 21
 22    protected SqsAsyncResponseOptions Options { get; }
 23    protected ILogger Logger { get; }
 24
 25    protected abstract string QueueName { get; }
 26    protected abstract SqsSubscriberOptions SubscriberOptions { get; }
 27    protected abstract SqsSubscriberRole SubscriberRole { get; }
 28    /// <summary>Handles the delivered message.</summary>
 29    protected abstract Task HandleMessageAsync(SqsTransportDelivery delivery, CancellationToken cancellationToken);
 30
 31    /// <summary>Runs this background operation until cancellation is requested.</summary>
 32    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
 33    {
 34        var queue = QueueName;
 35        SqsMessageDispatcher.ValidateOptions(Options, SubscriberOptions, SubscriberRole);
 36        var failures = 0;
 37
 38        while (!stoppingToken.IsCancellationRequested)
 39        {
 40            try
 41            {
 42                await RunSubscriberAsync(queue, stoppingToken).ConfigureAwait(false);
 43                return;
 44            }
 45            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
 46            {
 47                return;
 48            }
 49            catch (Exception ex) when (!stoppingToken.IsCancellationRequested)
 50            {
 51                failures++;
 52                var retryDelay = AsyncResponseRetry.Backoff(
 53                    failures,
 54                    Options.SubscriberRetryBaseDelay,
 55                    Options.SubscriberRetryMaxDelay);
 56                Logger.LogWarning(
 57                    ex,
 58                    "SQS subscriber failed for queue {Queue} ({Role}); retrying in {RetryDelay}.",
 59                    queue,
 60                    SubscriberRole,
 61                    retryDelay);
 62                await Task.Delay(retryDelay, stoppingToken).ConfigureAwait(false);
 63            }
 64        }
 65    }
 66
 67    private async Task RunSubscriberAsync(string queue, CancellationToken stoppingToken)
 68    {
 69        // A queue configured by name resolves through GetQueueUrl; failures here (queue not yet
 70        // provisioned, endpoint still starting) surface to the retry loop above.
 71        var queueUrl = SqsQueueAddress.IsUrl(queue)
 72            ? queue
 73            : await _client.GetQueueUrlAsync(queue, stoppingToken).ConfigureAwait(false);
 74
 75        await using var dispatcher = SqsMessageDispatcher.Create(
 76            HandleMessageAsync,
 77            Options,
 78            SubscriberOptions,
 79            Logger,
 80            queue,
 81            SubscriberRole);
 82
 83        Logger.LogInformation(
 84            "SQS subscriber started. Queue: {Queue}. Role: {Role}. AckMode: {AckMode}.",
 85            queue,
 86            SubscriberRole,
 87            SubscriberOptions.AckMode);
 88
 89        while (!stoppingToken.IsCancellationRequested)
 90        {
 91            // In early-ACK mode, receiving while the background queue is saturated would burn the
 92            // queue's redrive policy (SQS counts every receive), so wait for free capacity and never
 93            // request more messages than the dispatcher can accept.
 94            await dispatcher.WaitForCapacityAsync(stoppingToken).ConfigureAwait(false);
 95            var maxMessages = Math.Min(Options.MaxMessagesPerReceive, dispatcher.FreeCapacity);
 96
 97            var deliveries = await _client.ReceiveMessagesAsync(
 98                new SqsReceiveRequest(
 99                    queueUrl,
 100                    maxMessages,
 101                    Options.ReceiveWaitTime,
 102                    SubscriberOptions.VisibilityTimeout),
 103                stoppingToken).ConfigureAwait(false);
 104
 105            await DispatchBatchAsync(dispatcher, deliveries, queue, stoppingToken).ConfigureAwait(false);
 106        }
 107    }
 108
 109    private async Task DispatchBatchAsync(
 110        SqsMessageDispatcher dispatcher,
 111        IReadOnlyList<SqsTransportDelivery> deliveries,
 112        string queue,
 113        CancellationToken stoppingToken)
 114    {
 115        if (deliveries.Count == 0)
 116            return;
 117
 118        if (SubscriberOptions.AckMode is not SqsAckMode.AckAfterHandlerCompletes
 119            || SubscriberOptions.VisibilityRenewalInterval is not { } renewalInterval
 120            || SubscriberOptions.VisibilityTimeout is not { } visibilityTimeout)
 121        {
 122            foreach (var delivery in deliveries)
 123                await dispatcher.HandleAsync(delivery, stoppingToken).ConfigureAwait(false);
 124            return;
 125        }
 126
 127        // The batch is processed serially, so a slow handler lets the visibility timeout of the later
 128        // (still unprocessed) messages lapse and a competing consumer processes them a second time.
 129        // While the batch is in flight, a heartbeat resets every unsettled message's invisibility to
 130        // the configured visibility timeout.
 131        var progress = new BatchProgress(deliveries.Count);
 132        using var renewalCancellation = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken);
 133        var renewalTask = RenewVisibilityLoopAsync(
 134            deliveries,
 135            progress,
 136            renewalInterval,
 137            visibilityTimeout,
 138            queue,
 139            renewalCancellation.Token);
 140        try
 141        {
 142            for (var index = 0; index < deliveries.Count; index++)
 143            {
 144                var delivery = deliveries[index];
 145                var batchIndex = index;
 146                // The dispatcher's failure path shortens visibility to RedeliveryDelay while the
 147                // heartbeat still counts the message as unsettled (MarkSettled runs only after
 148                // HandleAsync returns). Routing the dispatcher's visibility changes through a
 149                // suppression mark — set before the change itself — keeps a racing heartbeat from
 150                // stretching that fast retry back out to the full visibility timeout.
 151                var tracked = delivery with
 152                {
 153                    ChangeVisibilityAsync = timeout =>
 154                    {
 155                        progress.SuppressRenewal(batchIndex);
 156                        return delivery.ChangeVisibilityAsync(timeout);
 157                    }
 158                };
 159                try
 160                {
 161                    await dispatcher.HandleAsync(tracked, stoppingToken).ConfigureAwait(false);
 162                }
 163                finally
 164                {
 165                    progress.MarkSettled();
 166                }
 167            }
 168        }
 169        finally
 170        {
 171            renewalCancellation.Cancel();
 172            await renewalTask.ConfigureAwait(false);
 173        }
 174    }
 175
 176    private async Task RenewVisibilityLoopAsync(
 177        IReadOnlyList<SqsTransportDelivery> deliveries,
 178        BatchProgress progress,
 179        TimeSpan renewalInterval,
 180        TimeSpan visibilityTimeout,
 181        string queue,
 182        CancellationToken cancellationToken)
 183    {
 184        try
 185        {
 186            while (true)
 187            {
 188                await Task.Delay(renewalInterval, cancellationToken).ConfigureAwait(false);
 189
 190                // Renew from the first unsettled message onward: that covers the message currently in
 191                // the handler plus everything still waiting its turn. Two settle paths race this
 192                // sweep, and only one of them is harmless. A handled message was deleted, so a late
 193                // renewal merely fails and is logged — SQS redelivery keeps at-least-once intact. A
 194                // failed message was NOT deleted (its receipt handle stays live) and already carries
 195                // the failure path's shortened RedeliveryDelay, so a late renewal here would SUCCEED
 196                // and stretch that fast retry back out to the full visibility timeout — the
 197                // suppression mark and the per-message re-read of the settled prefix keep the sweep
 198                // away from it.
 199                for (var i = progress.SettledCount; i < deliveries.Count; i++)
 200                {
 201                    if (i < progress.SettledCount || progress.IsRenewalSuppressed(i))
 202                        continue;
 203
 204                    var delivery = deliveries[i];
 205                    try
 206                    {
 207                        await delivery.ChangeVisibilityAsync(visibilityTimeout).ConfigureAwait(false);
 208                    }
 209                    catch (Exception ex)
 210                    {
 211                        Logger.LogWarning(
 212                            ex,
 213                            "Failed to renew visibility of SQS message {MessageId} on {Queue}; it may redeliver while st
 214                            delivery.MessageId,
 215                            queue);
 216                    }
 217                }
 218            }
 219        }
 220        catch (OperationCanceledException)
 221        {
 222            // The batch finished or the subscriber is stopping.
 223        }
 224    }
 225
 226    private sealed class BatchProgress
 227    {
 228        // One slot per batch message. Settled and suppressed only ever transition false→true, so
 229        // monotonic volatile writes/reads are enough — no lock, and a stale read only delays a
 230        // skip by one sweep pass.
 231        private readonly bool[] _renewalSuppressed;
 232        private int _settledCount;
 233
 2234        public BatchProgress(int batchSize) => _renewalSuppressed = new bool[batchSize];
 235
 2236        public int SettledCount => Volatile.Read(ref _settledCount);
 237
 2238        public void MarkSettled() => Interlocked.Increment(ref _settledCount);
 239
 240        /// <summary>Marks the message at <paramref name="index"/> as owning its own visibility; the renewal sweep must 
 2241        public void SuppressRenewal(int index) => Volatile.Write(ref _renewalSuppressed[index], true);
 242
 2243        public bool IsRenewalSuppressed(int index) => Volatile.Read(ref _renewalSuppressed[index]);
 244    }
 245}
 246
 247internal sealed class SqsWorkerSubscriber : SqsSubscriberService
 248{
 249    private readonly IAsyncResponseIngress _ingress;
 250
 251    /// <summary>Creates a worker subscriber for the configured SQS worker queue.</summary>
 252    public SqsWorkerSubscriber(
 253        IOptions<SqsAsyncResponseOptions> options,
 254        ISqsClient client,
 255        IAsyncResponseIngress ingress,
 256        ILogger<SqsWorkerSubscriber> logger)
 257        : base(options, client, logger)
 258    {
 259        _ingress = ingress;
 260    }
 261
 262    protected override string QueueName
 263        => SqsOptionsValidator.Required(Options.WorkerQueue, nameof(Options.WorkerQueue));
 264
 265    protected override SqsSubscriberOptions SubscriberOptions => Options.WorkerSubscriber;
 266    protected override SqsSubscriberRole SubscriberRole => SqsSubscriberRole.Worker;
 267
 268    /// <summary>Handles the delivered message.</summary>
 269    protected override Task HandleMessageAsync(SqsTransportDelivery delivery, CancellationToken cancellationToken)
 270        => _ingress.HandleWorkerMessageAsync(delivery.Body);
 271}
 272
 273internal sealed class SqsResponseIngressSubscriber : SqsSubscriberService
 274{
 275    private readonly IAsyncResponseIngress _ingress;
 276
 277    /// <summary>Creates a response subscriber for the configured SQS response queue.</summary>
 278    public SqsResponseIngressSubscriber(
 279        IOptions<SqsAsyncResponseOptions> options,
 280        ISqsClient client,
 281        IAsyncResponseIngress ingress,
 282        ILogger<SqsResponseIngressSubscriber> logger)
 283        : base(options, client, logger)
 284    {
 285        _ingress = ingress;
 286    }
 287
 288    protected override string QueueName
 289        => SqsOptionsValidator.Required(Options.ResponseQueue, nameof(Options.ResponseQueue));
 290
 291    protected override SqsSubscriberOptions SubscriberOptions => Options.ResponseSubscriber;
 292    protected override SqsSubscriberRole SubscriberRole => SqsSubscriberRole.ResponseIngress;
 293
 294    /// <summary>Handles the delivered message.</summary>
 295    protected override Task HandleMessageAsync(SqsTransportDelivery delivery, CancellationToken cancellationToken)
 296    {
 297        var correlationId = SqsCorrelationIdExtractor.Extract(delivery, delivery.Body, Options);
 298        return _ingress.HandleResponseMessageAsync(delivery.Body, correlationId);
 299    }
 300}
Generated by: ReportGenerator 5.5.10.0
08/03/2026 - 08:59:44
GitHub | reportgenerator.io

Methods/Properties

.ctor(int)
get_SettledCount()
MarkSettled()
SuppressRenewal(int)
IsRenewalSuppressed(int)