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

Information
Class: AsyncResponse.Transports.SQS.SqsWorkerTransport
Assembly: AsyncResponse.Transports.SQS
File(s): /_/src/Transports/AsyncResponse.Transports.SQS/SqsWorkerTransport.cs
Line coverage
92%
Covered lines: 103
Uncovered lines: 8
Coverable lines: 111
Total lines: 280
Line coverage: 92.7%
Branch coverage
78%
Covered branches: 60
Total branches: 76
Branch coverage: 78.9%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.cctor()100%11100%
.ctor(...)100%11100%
.ctor(...)100%11100%
.ctor(...)100%11100%
GetQueueUrlAsync()100%66100%
PublishAsync(...)100%11100%
get_MaxPublishDelay()50%22100%
get_MaxInFlightDuration()70%141066.66%
PublishAsync(...)50%11650%
PublishCoreAsync()83.33%242496.96%
ToMessageGroupId(...)50%22100%
IsValidMessageGroupId(...)57.14%211466.66%
SendWithRetryAsync()100%11100%
IsTransient(...)100%88100%
DisposeAsync()100%44100%

File(s)

/_/src/Transports/AsyncResponse.Transports.SQS/SqsWorkerTransport.cs

#LineLine coverage
 1using Amazon.SQS;
 2using Microsoft.Extensions.Options;
 3using System.Diagnostics;
 4using System.Net;
 5using System.Security.Cryptography;
 6using System.Text;
 7using System.Text.Json;
 8
 9namespace AsyncResponse.Transports.SQS;
 10
 11/// <summary>Publishes <see cref="WorkerJobEnvelope"/> messages to an AWS SQS queue.</summary>
 12/// <remarks>
 13/// The worker queue URL is resolved lazily (a queue configured by name goes through
 14/// <c>GetQueueUrl</c> once) and cached for the lifetime of the transport; a transient resolution
 15/// failure on the first publish is not cached, so the next publish retries. When the worker queue
 16/// is a FIFO queue (name or URL ending in <c>.fifo</c>), the correlation id becomes the
 17/// <c>MessageGroupId</c> so one flow's jobs stay ordered (an id SQS would reject there — longer
 18/// than 128 characters, or anything outside ASCII letters, digits and punctuation — is replaced by
 19/// a stable hash of itself), and every message carries a unique
 20/// <c>MessageDeduplicationId</c> so distinct jobs of the same flow are never deduplicated away.
 21/// </remarks>
 22public sealed class SqsWorkerTransport : IWorkerTransport, IDelayedWorkerTransport, IWorkerTransportInFlightLimit, IAsyn
 23{
 24    /// <summary>The SQS per-message <c>DelaySeconds</c> ceiling (15 minutes).</summary>
 525    internal static readonly TimeSpan SqsMaxDelay = TimeSpan.FromSeconds(900);
 26
 27    /// <summary>
 28    /// The SQS in-flight ceiling: a message stays invisible for at most 12 hours counted from the
 29    /// <c>ReceiveMessage</c> that delivered it. Extending the visibility timeout does not reset
 30    /// that clock, and a <c>ChangeMessageVisibility</c> that would cross it is rejected.
 31    /// https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-visibility-timeout.html
 32    /// </summary>
 533    internal static readonly TimeSpan SqsMaxInFlightDuration = TimeSpan.FromHours(12);
 34
 35    /// <summary>The SQS <c>MessageGroupId</c> length limit.</summary>
 36    private const int MaxMessageGroupIdLength = 128;
 37
 38    /// <summary>Marks a <c>MessageGroupId</c> derived by hashing a correlation id SQS would reject.</summary>
 39    private const string HashedMessageGroupIdPrefix = "sha256-";
 40
 41    private readonly SqsAsyncResponseOptions _options;
 42    private readonly ISqsClient _client;
 43    private readonly bool _disposeClient;
 44    private readonly bool _isFifoQueue;
 25645    private readonly SemaphoreSlim _queueUrlGate = new(1, 1);
 46    private string? _queueUrl;
 47    private int _disposeGate;
 48    private bool _disposed;
 49
 50    /// <summary>Creates a worker transport backed by a client built from the configured options.</summary>
 51    public SqsWorkerTransport(IOptions<SqsAsyncResponseOptions> options)
 252        : this(options, SqsClientFactory.Create(options.Value), disposeClient: true)
 53    {
 254    }
 55
 56    internal SqsWorkerTransport(
 57        IOptions<SqsAsyncResponseOptions> options,
 58        ISqsClient client)
 25459        : this(options, client, disposeClient: false)
 60    {
 22661    }
 62
 25663    private SqsWorkerTransport(
 25664        IOptions<SqsAsyncResponseOptions> options,
 25665        ISqsClient client,
 25666        bool disposeClient)
 67    {
 25668        _options = options.Value;
 25669        SqsOptionsValidator.ValidateCommon(_options);
 22870        _client = client;
 22871        _disposeClient = disposeClient;
 22872        _isFifoQueue = SqsQueueAddress.IsFifo(_options.WorkerQueue);
 22873    }
 74
 75    private async Task<string> GetQueueUrlAsync(CancellationToken cancellationToken)
 76    {
 44877        var queueUrl = Volatile.Read(ref _queueUrl);
 44878        if (queueUrl is not null)
 22379            return queueUrl;
 80
 22581        await _queueUrlGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 82        try
 83        {
 22584            ObjectDisposedException.ThrowIf(_disposed, this);
 22185            if (_queueUrl is not null)
 286                return _queueUrl;
 87
 88            // Assign only after the await succeeds, so a faulted resolution is not cached and the
 89            // next publish retries instead of reusing a permanently failed lookup.
 21990            var resolved = SqsQueueAddress.IsUrl(_options.WorkerQueue)
 21991                ? _options.WorkerQueue
 21992                : await _client.GetQueueUrlAsync(_options.WorkerQueue, cancellationToken).ConfigureAwait(false);
 21793            _queueUrl = resolved;
 21794            return resolved;
 95        }
 96        finally
 97        {
 22598            _queueUrlGate.Release();
 99        }
 442100    }
 101
 102    /// <summary>Publishes the supplied worker job.</summary>
 103    public Task PublishAsync(WorkerJobEnvelope job, CancellationToken cancellationToken = default)
 449104        => PublishCoreAsync(job, delay: null, cancellationToken);
 105
 106    /// <inheritdoc/>
 107    /// <remarks>
 108    /// SQS caps a single hop at 15 minutes (<c>DelaySeconds</c> ≤ 900); longer waits ride the
 109    /// <see cref="WorkerJobEnvelope.NotBeforeUtc"/> re-publish chain, 15 minutes per hop.
 110    /// A FIFO worker queue reports <see cref="TimeSpan.Zero"/>: SQS rejects per-message
 111    /// <c>DelaySeconds</c> on FIFO queues, and advertising a capability the publish would then
 112    /// throw on lets a durable flow persist itself as sleeping before the enqueue fails —
 113    /// stranding the run. Zero routes the engine to its in-process fallback instead.
 114    /// </remarks>
 5115    public TimeSpan MaxPublishDelay => _isFifoQueue ? TimeSpan.Zero : SqsMaxDelay;
 116
 117    /// <inheritdoc/>
 118    /// <remarks>
 119    /// SQS redelivers a message the moment its visibility lapses, however alive its handler is.
 120    /// With <see cref="SqsSubscriberOptions.VisibilityRenewalInterval"/> set, the worker
 121    /// subscriber keeps extending the visibility until the 12-hour SQS maximum, which nothing can
 122    /// extend past. Without renewal an explicit <see cref="SqsSubscriberOptions.VisibilityTimeout"/>
 123    /// is itself the ceiling and is reported as such; when that is unset too, the queue's own
 124    /// visibility timeout governs — a value this transport never reads — so only the 12-hour
 125    /// upper bound can be reported: set <c>VisibilityTimeout</c> to advertise the real one.
 126    /// <c>null</c> in <see cref="SqsAckMode.AckAfterEnqueue"/>, where the message is deleted before
 127    /// its handler runs and nothing stays in flight at the broker.
 128    /// </remarks>
 129    public TimeSpan? MaxInFlightDuration
 240130        => _options.WorkerSubscriber switch
 240131        {
 0132            { AckMode: SqsAckMode.AckAfterEnqueue } => null,
 240133            // The positivity guard covers a publisher-only process, where the subscriber options
 240134            // are never validated because no subscriber starts.
 0135            { VisibilityRenewalInterval: null, VisibilityTimeout: { } visibilityTimeout } when visibilityTimeout > TimeS
 0136                => visibilityTimeout,
 240137            _ => SqsMaxInFlightDuration
 240138        };
 139
 140    /// <inheritdoc/>
 141    public Task PublishAsync(WorkerJobEnvelope job, TimeSpan delay, CancellationToken cancellationToken = default)
 142    {
 1143        if (_isFifoQueue && delay > TimeSpan.Zero)
 144        {
 145            // SQS rejects per-message DelaySeconds on FIFO queues (queue-level delay only), and a
 146            // silently dropped delay would break every due-time above. Fail loudly with the way out.
 0147            throw new InvalidOperationException(
 0148                $"SQS FIFO queues do not support per-message delays, so delayed worker jobs (and suspended durable-flow 
 0149                "Use a standard worker queue for delayed delivery, or keep timers in process by leaving the transport wi
 150        }
 151
 1152        ArgumentOutOfRangeException.ThrowIfGreaterThan(delay, MaxPublishDelay);
 1153        return PublishCoreAsync(job, delay > TimeSpan.Zero ? delay : null, cancellationToken);
 154    }
 155
 156    private async Task PublishCoreAsync(WorkerJobEnvelope job, TimeSpan? delay, CancellationToken cancellationToken)
 157    {
 450158        ArgumentNullException.ThrowIfNull(job);
 159
 448160        using var activity = AsyncResponseDiagnostics.StartActivity(
 448161            "asyncresponse.worker.publish",
 448162            ActivityKind.Producer,
 448163            job.CorrelationId);
 448164        activity?.SetTag("asyncresponse.transport", "aws_sqs");
 448165        activity?.SetTag("messaging.system", "aws_sqs");
 448166        activity?.SetTag("messaging.destination.name", _options.WorkerQueue);
 448167        AsyncResponseDiagnostics.SetReplyTarget(activity, job.ReplyTarget);
 448168        AsyncResponseDiagnostics.SetWorker(activity, job.Call);
 169
 170        try
 171        {
 448172            var messageAttributes = new Dictionary<string, string>(StringComparer.Ordinal);
 448173            if (!string.IsNullOrWhiteSpace(job.CorrelationId))
 125174                messageAttributes[_options.CorrelationIdAttribute] = job.CorrelationId;
 175
 448176            var queueUrl = await GetQueueUrlAsync(cancellationToken).ConfigureAwait(false);
 442177            var message = new SqsOutboundMessage(
 442178                queueUrl,
 442179                AsyncResponseJson.Serialize(job),
 442180                string.IsNullOrWhiteSpace(job.CorrelationId) ? null : job.CorrelationId,
 442181                MessageGroupId: _isFifoQueue
 442182                    ? (string.IsNullOrWhiteSpace(job.CorrelationId) ? _options.FifoMessageGroupIdFallback : ToMessageGro
 442183                    : null,
 442184                MessageDeduplicationId: _isFifoQueue ? Guid.NewGuid().ToString("N") : null,
 442185                messageAttributes,
 442186                DelaySeconds: delay is { } pending ? (int)Math.Ceiling(pending.TotalSeconds) : null);
 442187            if (delay is { } delayTag)
 1188                activity?.SetTag("asyncresponse.worker.delay_seconds", delayTag.TotalSeconds);
 189
 442190            var messageId = await SendWithRetryAsync(message, cancellationToken).ConfigureAwait(false);
 436191            activity?.SetTag("messaging.message.id", messageId);
 436192        }
 12193        catch (Exception ex)
 194        {
 12195            AsyncResponseDiagnostics.SetError(activity, ex);
 12196            throw;
 197        }
 436198    }
 199
 200    /// <summary>
 201    /// Maps a correlation id onto a <c>MessageGroupId</c> SQS accepts. A correlation id is portable
 202    /// text — spaces, non-ASCII, up to 400 characters — while SQS allows at most 128 characters of
 203    /// ASCII letters, digits and punctuation there and rejects the whole <c>SendMessage</c>
 204    /// otherwise, so every FIFO publish for such an id failed. Conforming ids pass through
 205    /// unchanged (existing groups keep their ordering); the rest become a stable SHA-256 of the id,
 206    /// so one id still always lands in one group.
 207    /// </summary>
 208    internal static string ToMessageGroupId(string correlationId)
 4209        => IsValidMessageGroupId(correlationId)
 4210            ? correlationId
 4211            // Uppercase hex: ToHexStringLower is .NET 9+, and this package still targets net8.0.
 4212            : HashedMessageGroupIdPrefix + Convert.ToHexString(SHA256.HashData(Encoding.UTF8.GetBytes(correlationId)));
 213
 214    /// <summary>Whether SQS accepts the value as a <c>MessageGroupId</c> (or <c>MessageDeduplicationId</c>).</summary>
 215    internal static bool IsValidMessageGroupId(string value)
 216    {
 4217        if (value.Length is 0 or > MaxMessageGroupIdLength)
 0218            return false;
 219
 80220        foreach (var character in value)
 221        {
 222            // ASCII letters, digits and punctuation: exactly the printable range minus the space.
 36223            if (character is <= ' ' or > '~')
 0224                return false;
 225        }
 226
 4227        return true;
 228    }
 229
 230    private async Task<string> SendWithRetryAsync(
 231        SqsOutboundMessage message,
 232        CancellationToken cancellationToken)
 233    {
 448234        for (var attempt = 1; ; attempt++)
 235        {
 236            try
 237            {
 448238                return await _client.SendMessageAsync(message, cancellationToken).ConfigureAwait(false);
 239            }
 12240            catch (Exception ex) when (IsTransient(ex) && attempt < _options.PublishMaxAttempts && !cancellationToken.Is
 241            {
 6242                var delay = AsyncResponseRetry.Backoff(attempt, _options.PublishRetryBaseDelay, _options.PublishRetryMax
 6243                await Task.Delay(delay, cancellationToken).ConfigureAwait(false);
 244            }
 245        }
 436246    }
 247
 248    /// <summary>Classifies AWS SQS send failures worth retrying at the transport level.</summary>
 249    internal static bool IsTransient(Exception exception)
 22250        => exception is AmazonSQSException sqsException
 22251            && (sqsException.Retryable is not null
 22252                || sqsException.StatusCode >= HttpStatusCode.InternalServerError
 22253                || string.Equals(sqsException.ErrorCode, "RequestThrottled", StringComparison.Ordinal)
 22254                || string.Equals(sqsException.ErrorCode, "ThrottlingException", StringComparison.Ordinal));
 255
 256    /// <summary>Releases resources held by this instance.</summary>
 257    public async ValueTask DisposeAsync()
 258    {
 402259        if (Interlocked.Exchange(ref _disposeGate, 1) != 0)
 198260            return;
 261
 204262        await _queueUrlGate.WaitAsync().ConfigureAwait(false);
 263        try
 264        {
 204265            _disposed = true;
 204266            if (_disposeClient)
 2267                await _client.DisposeAsync().ConfigureAwait(false);
 204268        }
 269        finally
 270        {
 271            // Release, never Dispose: SemaphoreSlim.Dispose does not complete pending WaitAsync
 272            // waiters, so disposing here would strand publishers parked on the gate forever (and
 273            // the first woken waiter's finally would throw trying to Release a disposed
 274            // semaphore, never handing the permit on). Released, each parked waiter wakes in
 275            // turn and observes _disposed; the gate holds no unmanaged resources, so leaving it
 276            // undisposed leaks nothing.
 204277            _queueUrlGate.Release();
 278        }
 402279    }
 280}
Generated by: ReportGenerator 5.5.11.0
09/22/2026 - 09:41:04
GitHub | reportgenerator.io

Methods/Properties

.cctor()
.ctor(Microsoft.Extensions.Options.IOptions`1<AsyncResponse.Transports.SQS.SqsAsyncResponseOptions>,AsyncResponse.Transports.SQS.ISqsClient,System.Boolean)
.ctor(Microsoft.Extensions.Options.IOptions`1<AsyncResponse.Transports.SQS.SqsAsyncResponseOptions>)
.ctor(Microsoft.Extensions.Options.IOptions`1<AsyncResponse.Transports.SQS.SqsAsyncResponseOptions>,AsyncResponse.Transports.SQS.ISqsClient)
GetQueueUrlAsync()
PublishAsync(AsyncResponse.WorkerJobEnvelope,System.Threading.CancellationToken)
get_MaxPublishDelay()
get_MaxInFlightDuration()
PublishAsync(AsyncResponse.WorkerJobEnvelope,System.TimeSpan,System.Threading.CancellationToken)
PublishCoreAsync()
ToMessageGroupId(System.String)
IsValidMessageGroupId(System.String)
SendWithRetryAsync()
IsTransient(System.Exception)
DisposeAsync()