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

Information
Class: AsyncResponse.Transports.DbMessageDispatcherBase
Assembly: AsyncResponse.Transports.SqlServer
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/Transports/Shared/DbTransportShared.cs
Line coverage
98%
Covered lines: 156
Uncovered lines: 3
Coverable lines: 159
Total lines: 428
Line coverage: 98.1%
Branch coverage
100%
Covered branches: 36
Total branches: 36
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%44100%
HandleAsync()100%22100%
RenewLeaseLoopAsync()100%22100%
ExecuteHandlerAsync()100%1616100%
HandleEarlyAckAsync()100%22100%
BackgroundWorkerLoopAsync()100%11100%
HandleFailureAsync()100%66100%
InvokeBackgroundFailureAsync()100%22100%
DisposeAsync()100%2276.47%

File(s)

/home/runner/work/AsyncResponse/AsyncResponse/src/Transports/Shared/DbTransportShared.cs

#LineLine coverage
 1using Microsoft.Extensions.Logging;
 2using System.Text.Json;
 3using System.Text.Json.Nodes;
 4using System.Threading.Channels;
 5
 6namespace AsyncResponse.Transports;
 7
 8// Shared source for the database-backed worker transports (PostgreSQL, SQL Server, MongoDB),
 9// mirroring src/Channels/Shared/DbChannelShared.cs: each transport csproj pulls this file in via
 10// <Compile Include="..\Shared\DbTransportShared.cs" />, so the base class compiles INTO each
 11// provider assembly against that provider's concrete seam types. The seam is bound per project
 12// with global using aliases (declared at the top of the provider's MessageDispatcher file):
 13//
 14//   DbTransportOptions          -> the provider's transport options (e.g. PostgreSqlAsyncResponseTransportOptions)
 15//   DbSubscriberOptions         -> the provider's subscriber options (e.g. PostgreSqlSubscriberOptions)
 16//   DbTransportDelivery         -> the provider's claimed-delivery type (e.g. PostgreSqlTransportDelivery)
 17//   DbSubscriberRole            -> the provider's subscriber-role enum
 18//   DbAckMode                   -> the provider's ack-mode enum
 19//   DbTransportOptionsValidator -> the provider's static options validator
 20//   DbBackgroundFailureContext  -> the provider's OnBackgroundFailure context type
 21//
 22// Because the aliases resolve to concrete sealed types at compile time, delivery calls stay
 23// direct — no interface dispatch on the per-message path. The only provider-specific inputs are
 24// three display strings supplied by the derived constructor: the provider name rendered into log
 25// messages, the queue-item noun ("row"/"document"), and the lowercase telemetry tag. Rendered log
 26// output and activity tags are byte-identical to the pre-extraction per-provider sources.
 27
 28/// <summary>
 29/// Applies acknowledgement, redelivery, and dead-letter policy to database transport deliveries:
 30/// ack-after-handler with fenced lease renewal, opt-in early ACK behind a bounded in-process
 31/// queue with drain-on-dispose, attempt-capped dead-lettering, and the consumer receive span.
 32/// Derived dispatchers supply only the provider display name, queue-item noun, and telemetry tag.
 33/// </summary>
 34internal abstract class DbMessageDispatcherBase : IAsyncDisposable
 35{
 36    private readonly Func<DbTransportDelivery, CancellationToken, Task> _handler;
 37    private readonly DbTransportOptions _options;
 38    private readonly DbSubscriberOptions _subscriberOptions;
 39    private readonly ILogger _logger;
 40    private readonly DbSubscriberRole _role;
 41    private readonly string _providerName;
 42    private readonly string _unitNoun;
 43    private readonly string _receiveActivityName;
 44    private readonly string _transportTag;
 45    private readonly string _roleTagName;
 46    private readonly string _ackModeTagName;
 47
 48    private readonly Channel<DbTransportDelivery>? _backgroundQueue;
 49    private readonly Task[]? _backgroundWorkers;
 50    private readonly CancellationTokenSource? _backgroundCts;
 51
 352    protected DbMessageDispatcherBase(
 353        Func<DbTransportDelivery, CancellationToken, Task> handler,
 354        DbTransportOptions options,
 355        DbSubscriberOptions subscriberOptions,
 356        ILogger logger,
 357        DbSubscriberRole role,
 358        string providerName,
 359        string unitNoun,
 360        string telemetryName)
 61    {
 362        DbTransportOptionsValidator.ValidateSubscriber(options, subscriberOptions, role.ToString());
 63
 364        _handler = handler;
 365        _options = options;
 366        _subscriberOptions = subscriberOptions;
 367        _logger = logger;
 368        _role = role;
 369        _providerName = providerName;
 370        _unitNoun = unitNoun;
 371        _receiveActivityName = $"asyncresponse.{telemetryName}.receive";
 372        _transportTag = telemetryName;
 373        _roleTagName = $"asyncresponse.{telemetryName}.role";
 374        _ackModeTagName = $"asyncresponse.{telemetryName}.ack_mode";
 75
 376        if (subscriberOptions.AckMode is DbAckMode.AckAfterEnqueue)
 77        {
 378            _backgroundQueue = Channel.CreateBounded<DbTransportDelivery>(new BoundedChannelOptions(subscriberOptions.Ba
 379            {
 380                SingleReader = false,
 381                SingleWriter = true,
 382                FullMode = BoundedChannelFullMode.Wait
 383            });
 384            _backgroundCts = new CancellationTokenSource();
 385            _backgroundWorkers = new Task[subscriberOptions.BackgroundWorkerCount];
 386            for (var i = 0; i < _backgroundWorkers.Length; i++)
 387                _backgroundWorkers[i] = Task.Run(() => BackgroundWorkerLoopAsync(_backgroundCts.Token));
 88        }
 389    }
 90
 91    /// <summary>Handles one claimed queue item.</summary>
 92    public async Task HandleAsync(DbTransportDelivery delivery, CancellationToken cancellationToken)
 93    {
 394        if (_subscriberOptions.AckMode is DbAckMode.AckAfterEnqueue)
 95        {
 396            await HandleEarlyAckAsync(delivery).ConfigureAwait(false);
 397            return;
 98        }
 99
 100        try
 101        {
 102            // While the handler runs, a fenced heartbeat keeps extending the claim's lease at
 103            // LockTimeout/2 cadence so a slow handler does not let the lock lapse and a competing
 104            // subscriber re-claim (and duplicate-process) the queue item.
 3105            using var renewalCancellation = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
 3106            var renewalTask = RenewLeaseLoopAsync(delivery, renewalCancellation.Token);
 107            try
 108            {
 3109                await ExecuteHandlerAsync(delivery, cancellationToken).ConfigureAwait(false);
 110            }
 111            finally
 112            {
 3113                renewalCancellation.Cancel();
 3114                await renewalTask.ConfigureAwait(false);
 115            }
 116
 3117            await delivery.AckAsync().ConfigureAwait(false);
 3118        }
 3119        catch (Exception ex)
 120        {
 3121            await HandleFailureAsync(delivery, ex, cancellationToken).ConfigureAwait(false);
 122        }
 3123    }
 124
 125    private async Task RenewLeaseLoopAsync(DbTransportDelivery delivery, CancellationToken cancellationToken)
 126    {
 3127        var interval = TimeSpan.FromTicks(Math.Max(1, _options.LockTimeout.Ticks / 2));
 128        try
 129        {
 130            while (true)
 131            {
 3132                await Task.Delay(interval, cancellationToken).ConfigureAwait(false);
 133
 134                bool renewed;
 135                try
 136                {
 2137                    renewed = await delivery.RenewAsync().ConfigureAwait(false);
 2138                }
 2139                catch (Exception ex)
 140                {
 2141                    _logger.LogWarning(
 2142                        ex,
 2143                        "Failed to renew the lease of {Provider} message {MessageId} on queue {Queue} ({Role}); retrying
 2144                        _providerName,
 2145                        delivery.Id,
 2146                        delivery.Queue,
 2147                        _role);
 2148                    continue;
 149                }
 150
 2151                if (!renewed)
 152                {
 153                    // The lock_id fence no longer matches: the lease expired and another subscriber
 154                    // claimed the queue item. Stop renewing; the fenced ack/NAK will no-op for this
 155                    // claim.
 2156                    _logger.LogWarning(
 2157                        "Lease of {Provider} message {MessageId} on queue {Queue} ({Role}) was lost; another subscriber 
 2158                        _providerName,
 2159                        delivery.Id,
 2160                        delivery.Queue,
 2161                        _role);
 3162                    return;
 163                }
 164            }
 165        }
 3166        catch (OperationCanceledException)
 167        {
 168            // The handler finished or the subscriber is stopping.
 3169        }
 3170    }
 171
 172    // Single choke point for handler execution so both ACK modes emit the consumer receive span.
 173    private async Task ExecuteHandlerAsync(DbTransportDelivery delivery, CancellationToken cancellationToken)
 174    {
 3175        using var activity = AsyncResponseDiagnostics.StartActivity(
 3176            _receiveActivityName,
 3177            System.Diagnostics.ActivityKind.Consumer);
 3178        activity?.SetTag("asyncresponse.transport", _transportTag);
 3179        activity?.SetTag(_roleTagName, _role.ToString());
 3180        activity?.SetTag(_ackModeTagName, _subscriberOptions.AckMode.ToString());
 3181        activity?.SetTag("messaging.system", _transportTag);
 3182        activity?.SetTag("messaging.destination.name", delivery.Queue);
 3183        activity?.SetTag("messaging.message.id", delivery.Id.ToString());
 3184        activity?.SetTag("messaging.message.delivery_attempt", delivery.Attempt);
 185
 3186        if (delivery.Headers.TryGetValue(_options.CorrelationIdHeader, out var correlationId))
 3187            AsyncResponseDiagnostics.SetCorrelationId(activity, correlationId);
 188
 189        try
 190        {
 3191            await _handler(delivery, cancellationToken).ConfigureAwait(false);
 3192        }
 3193        catch (Exception ex)
 194        {
 3195            AsyncResponseDiagnostics.SetError(activity, ex);
 3196            throw;
 197        }
 3198    }
 199
 200    private async Task HandleEarlyAckAsync(DbTransportDelivery delivery)
 201    {
 3202        if (_backgroundQueue!.Writer.TryWrite(delivery))
 203        {
 3204            await delivery.AckAsync().ConfigureAwait(false);
 205        }
 206        else
 207        {
 2208            _logger.LogDebug("Background queue full for {Provider} {Role}; releasing {Unit} for redelivery.", _providerN
 2209            await delivery.NakAsync(_subscriberOptions.RedeliveryDelay).ConfigureAwait(false);
 210        }
 3211    }
 212
 213    private async Task BackgroundWorkerLoopAsync(CancellationToken cancellationToken)
 214    {
 215        // Token-less ReadAllAsync: on shutdown the queue is completed and fully drained, so every
 216        // already-ACKed queue item is attempted (with the drain token once the drain budget lapses)
 217        // instead of being silently dropped; each failure is dead-lettered and surfaced below.
 3218        await foreach (var delivery in _backgroundQueue!.Reader.ReadAllAsync().ConfigureAwait(false))
 219        {
 220            try
 221            {
 3222                await ExecuteHandlerAsync(delivery, cancellationToken).ConfigureAwait(false);
 3223            }
 3224            catch (Exception ex)
 225            {
 2226                _logger.LogError(ex, "{Provider} background handler failed for {Role} on queue {Queue} after early ACK."
 2227                if (!await delivery.DeadLetterAsync(ex, false, CancellationToken.None).ConfigureAwait(false))
 228                {
 2229                    _logger.LogError(
 2230                        "Failed to dead-letter already-ACKed {Provider} message {MessageId} on queue {Queue} ({Role}); t
 2231                        _providerName,
 2232                        delivery.Id,
 2233                        delivery.Queue,
 2234                        _role);
 235                }
 236
 2237                await InvokeBackgroundFailureAsync(delivery, ex).ConfigureAwait(false);
 3238            }
 3239        }
 3240    }
 241
 242    private async Task HandleFailureAsync(DbTransportDelivery delivery, Exception exception, CancellationToken cancellat
 243    {
 2244        var maxAttempts = _subscriberOptions.MaxDeliveryAttempts;
 2245        if (maxAttempts > 0 && delivery.Attempt >= maxAttempts)
 246        {
 2247            _logger.LogError(
 2248                exception,
 2249                "{Provider} message on queue {Queue} ({Role}) failed after {Attempts} attempts; dead-lettering.",
 2250                _providerName,
 2251                delivery.Queue,
 2252                _role,
 2253                delivery.Attempt);
 254
 2255            var deadLettered = await delivery.DeadLetterAsync(exception, true, cancellationToken).ConfigureAwait(false);
 2256            if (!deadLettered)
 257            {
 2258                _logger.LogWarning(exception, "{Provider} dead-letter publish failed for queue {Queue} ({Role}); releasi
 2259                await delivery.NakAsync(_subscriberOptions.RedeliveryDelay).ConfigureAwait(false);
 260            }
 261        }
 262        else
 263        {
 2264            _logger.LogWarning(
 2265                exception,
 2266                "{Provider} message on queue {Queue} ({Role}) failed on attempt {Attempt}; releasing for redelivery.",
 2267                _providerName,
 2268                delivery.Queue,
 2269                _role,
 2270                delivery.Attempt);
 2271            await delivery.NakAsync(_subscriberOptions.RedeliveryDelay).ConfigureAwait(false);
 272        }
 2273    }
 274
 275    private async Task InvokeBackgroundFailureAsync(DbTransportDelivery delivery, Exception exception)
 276    {
 2277        if (_subscriberOptions.OnBackgroundFailure is null)
 2278            return;
 279
 280        try
 281        {
 2282            delivery.Headers.TryGetValue(_options.CorrelationIdHeader, out var correlationId);
 2283            var context = new DbBackgroundFailureContext(delivery.Queue, _role.ToString(), delivery.Attempt, correlation
 2284            await _subscriberOptions.OnBackgroundFailure(context).ConfigureAwait(false);
 2285        }
 2286        catch (Exception ex)
 287        {
 2288            _logger.LogError(ex, "{Provider} OnBackgroundFailure callback threw for {Role}.", _providerName, _role);
 2289        }
 2290    }
 291
 292    /// <inheritdoc />
 293    public async ValueTask DisposeAsync()
 294    {
 3295        if (_backgroundQueue is null)
 3296            return;
 297
 3298        _backgroundQueue.Writer.TryComplete();
 299        try
 300        {
 3301            await Task.WhenAll(_backgroundWorkers!).WaitAsync(_subscriberOptions.BackgroundDrainTimeout).ConfigureAwait(
 3302            _backgroundCts!.Dispose();
 3303        }
 304        catch (TimeoutException)
 305        {
 2306            _logger.LogWarning("{Provider} background handlers for {Role} did not drain within {Timeout}.", _providerNam
 2307            await _backgroundCts!.CancelAsync().ConfigureAwait(false);
 308
 309            // The workers are still running and observe _backgroundCts.Token inside ReadAllAsync, so disposing
 310            // it now would throw ObjectDisposedException inside them. Dispose once they actually finish, off
 311            // the shutdown path, so the source is not leaked either.
 3312            _ = Task.WhenAll(_backgroundWorkers!).ContinueWith(
 3313                _ => _backgroundCts.Dispose(),
 3314                CancellationToken.None,
 3315                TaskContinuationOptions.ExecuteSynchronously,
 3316                TaskScheduler.Default);
 317        }
 0318        catch (Exception ex)
 319        {
 320            // WhenAll only completes once every worker has finished, so the source is safe to dispose here.
 0321            _logger.LogDebug(ex, "{Provider} background worker drain for {Role} ended with an error.", _providerName, _r
 0322            _backgroundCts!.Dispose();
 1323        }
 3324    }
 325}
 326
 327/// <summary>
 328/// Extracts the AsyncResponse correlation id from the queue item's metadata first, then from the
 329/// JSON response body via configured paths. Shared verbatim by the three database transports —
 330/// the header name and JSON paths both come from the aliased options type.
 331/// </summary>
 332internal static class DbCorrelationIdExtractor
 333{
 334    public static string? Extract(
 335        IReadOnlyDictionary<string, string>? headers,
 336        string messageJson,
 337        DbTransportOptions options)
 338    {
 339        var headerName = DbTransportOptionsValidator.Required(options.CorrelationIdHeader, nameof(options.CorrelationIdH
 340        if (headers is not null && headers.TryGetValue(headerName, out var headerValue) && !string.IsNullOrWhiteSpace(he
 341            return headerValue;
 342
 343        var jsonPaths = options.CorrelationIdJsonPaths;
 344        if (jsonPaths is null || jsonPaths.Length == 0 || string.IsNullOrWhiteSpace(messageJson))
 345            return null;
 346
 347        JsonNode? root;
 348        try
 349        {
 350            root = JsonNode.Parse(messageJson);
 351        }
 352        catch (JsonException)
 353        {
 354            return null;
 355        }
 356
 357        if (root is null)
 358            return null;
 359
 360        foreach (var path in jsonPaths)
 361        {
 362            var value = TryReadPath(root, path);
 363            if (!string.IsNullOrWhiteSpace(value))
 364                return value;
 365        }
 366
 367        return null;
 368    }
 369
 370    private static string? TryReadPath(JsonNode root, string path)
 371    {
 372        if (string.IsNullOrWhiteSpace(path))
 373            return null;
 374
 375        var current = root;
 376        foreach (var segment in path.Split('.', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries))
 377        {
 378            current = UnwrapJsonString(current);
 379            if (current is not JsonObject obj)
 380                return null;
 381
 382            current = TryGetProperty(obj, segment);
 383            if (current is null)
 384                return null;
 385        }
 386
 387        current = UnwrapJsonString(current);
 388        return current switch
 389        {
 390            JsonValue value when value.TryGetValue<string>(out var s) => s,
 391            JsonValue value => value.ToString(),
 392            _ => null
 393        };
 394    }
 395
 396    private static JsonNode? TryGetProperty(JsonObject obj, string name)
 397    {
 398        if (obj.TryGetPropertyValue(name, out var exact))
 399            return exact;
 400
 401        foreach (var property in obj)
 402        {
 403            if (string.Equals(property.Key, name, StringComparison.OrdinalIgnoreCase))
 404                return property.Value;
 405        }
 406
 407        return null;
 408    }
 409
 410    private static JsonNode? UnwrapJsonString(JsonNode? node)
 411    {
 412        if (node is not JsonValue value || !value.TryGetValue<string>(out var text))
 413            return node;
 414
 415        var trimmed = text.AsSpan().TrimStart();
 416        if (trimmed.Length == 0 || (trimmed[0] != '{' && trimmed[0] != '['))
 417            return node;
 418
 419        try
 420        {
 421            return JsonNode.Parse(text);
 422        }
 423        catch (JsonException)
 424        {
 425            return node;
 426        }
 427    }
 428}