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

Information
Class: AsyncResponse.Transports.SqlServer.SqlServerTransportDelivery
Assembly: AsyncResponse.Transports.SqlServer
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.SqlServer/SqlServerTransportStore.cs
Line coverage
100%
Covered lines: 10
Uncovered lines: 0
Coverable lines: 10
Total lines: 462
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%

File(s)

/home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.SqlServer/SqlServerTransportStore.cs

#LineLine coverage
 1using Microsoft.Data.SqlClient;
 2using Microsoft.Extensions.Logging;
 3using Microsoft.Extensions.Options;
 4using System.Runtime.CompilerServices;
 5using System.Text.Json;
 6
 7namespace AsyncResponse.Transports.SqlServer;
 8
 9internal enum SqlServerSubscriberRole
 10{
 11    Worker,
 12    ResponseIngress
 13}
 14
 15/// <summary>A claimed SQL Server transport row, decoupled from SqlClient types for dispatch tests.</summary>
 16/// <remarks>
 17/// <c>RenewAsync</c> extends the claim's lease (<c>locked_until</c>) by the original lock timeout,
 18/// fenced on the claim's <c>lock_id</c>; it returns <c>false</c> when the fence no longer matches
 19/// (the lease lapsed and another subscriber re-claimed the row).
 20/// </remarks>
 321internal sealed record SqlServerTransportDelivery(
 322    Guid Id,
 323    string Queue,
 324    string Payload,
 325    IReadOnlyDictionary<string, string> Headers,
 326    int Attempt,
 327    Func<ValueTask> AckAsync,
 328    Func<TimeSpan, ValueTask> NakAsync,
 329    Func<Exception, bool, CancellationToken, ValueTask<bool>> DeadLetterAsync,
 330    Func<ValueTask<bool>> RenewAsync);
 31
 32/// <summary>Small SQL adapter for the SQL Server transport queue table.</summary>
 33internal sealed class SqlServerTransportStore
 34{
 35    // SQL Server duplicate-key error numbers: 2627 = PRIMARY KEY/UNIQUE constraint violation,
 36    // 2601 = unique index violation. A concurrent retry of the same idempotent publish can lose the
 37    // WHERE NOT EXISTS race; the duplicate is the outcome the caller asked for, not a failure.
 38    private const int PrimaryKeyViolation = 2627;
 39    private const int UniqueIndexViolation = 2601;
 40
 41    // Interpolated into DDL: literal braces cannot appear directly inside the interpolated raw string.
 42    private const string EmptyJsonObject = "{}";
 43
 44    private readonly string _connectionString;
 45    private readonly SqlServerAsyncResponseTransportOptions _options;
 46    private readonly ILogger<SqlServerTransportStore>? _logger;
 47    private readonly SemaphoreSlim _ensureGate = new(1, 1);
 48    private bool _created;
 49    private long _lastDeadLetterPruneTicks;
 50
 51    public SqlServerTransportStore(
 52        IOptions<SqlServerAsyncResponseTransportOptions> options,
 53        ILogger<SqlServerTransportStore>? logger = null)
 54    {
 55        _options = options.Value;
 56        _logger = logger;
 57        SqlServerTransportOptionsValidator.ValidateCommon(_options);
 58        _connectionString = _options.ConnectionString!;
 59        Schema = Quote(_options.SchemaName);
 60        MessageTable = $"{Schema}.{Quote(_options.MessageTable)}";
 61    }
 62
 63    public string Schema { get; }
 64    public string MessageTable { get; }
 65
 66    /// <summary>
 67    /// Raised after a row is inserted (with the logical queue name) or released for retry
 68    /// (<c>null</c>). Same-process subscribers use it to wake immediately instead of waiting out
 69    /// their empty-poll delay; SQL Server has no LISTEN/NOTIFY, so cross-process wakes rely on polling.
 70    /// </summary>
 71    public event Action<string?>? MessagePublished;
 72
 73    public async Task EnsureCreatedAsync(CancellationToken cancellationToken = default)
 74    {
 75        if (_created || !_options.AutoCreateSchema)
 76            return;
 77
 78        await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 79        try
 80        {
 81            if (_created)
 82                return;
 83
 84            await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 85            await using var transaction = (SqlTransaction)await connection.BeginTransactionAsync(cancellationToken).Conf
 86
 87            // Serialize schema creation across processes. The IF-NOT-EXISTS guards are not atomic
 88            // against a concurrent create of the same object: two instances starting together both
 89            // pass the existence check and collide on the catalog (error 2714/2627). A
 90            // transaction-scoped application lock (keyed by schema, shared with the channel store)
 91            // lets one instance build the schema while the rest wait and then find it already present.
 92            await using (var lockCommand = connection.CreateCommand())
 93            {
 94                lockCommand.Transaction = transaction;
 95                lockCommand.CommandText =
 96                    """
 97                    DECLARE @lock_result int;
 98                    EXEC @lock_result = sp_getapplock
 99                        @Resource = @lock_resource,
 100                        @LockMode = 'Exclusive',
 101                        @LockOwner = 'Transaction',
 102                        @LockTimeout = 60000;
 103                    IF @lock_result < 0
 104                        THROW 51000, N'Failed to acquire the AsyncResponse DDL application lock.', 1;
 105                    """;
 106                lockCommand.Parameters.AddWithValue("@lock_resource", SchemaLockResource(_options.SchemaName));
 107                await lockCommand.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 108            }
 109
 110            await using var command = connection.CreateCommand();
 111            command.Transaction = transaction;
 112            command.CommandText =
 113                $"""
 114                IF SCHEMA_ID(N'{_options.SchemaName}') IS NULL
 115                    EXEC(N'CREATE SCHEMA {Schema}');
 116
 117                IF OBJECT_ID(N'{MessageTable}', N'U') IS NULL
 118                CREATE TABLE {MessageTable} (
 119                    id uniqueidentifier NOT NULL PRIMARY KEY NONCLUSTERED,
 120                    queue nvarchar(200) NOT NULL,
 121                    payload_json nvarchar(max) NOT NULL,
 122                    headers_json nvarchar(max) NOT NULL DEFAULT N'{EmptyJsonObject}',
 123                    created_at datetime2 NOT NULL DEFAULT SYSUTCDATETIME(),
 124                    available_at datetime2 NOT NULL DEFAULT SYSUTCDATETIME(),
 125                    locked_until datetime2 NULL,
 126                    lock_id uniqueidentifier NULL,
 127                    attempts int NOT NULL DEFAULT 0,
 128                    dead_letter_reason nvarchar(max) NULL
 129                );
 130
 131                IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name = N'{IndexName(_options.MessageTable, "claim")}' AND
 132                    CREATE INDEX {Quote(IndexName(_options.MessageTable, "claim"))}
 133                        ON {MessageTable} (queue, available_at, locked_until, created_at);
 134                IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name = N'{IndexName(_options.MessageTable, "created")}' A
 135                    CREATE INDEX {Quote(IndexName(_options.MessageTable, "created"))}
 136                        ON {MessageTable} (created_at);
 137                """;
 138            await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 139            await transaction.CommitAsync(cancellationToken).ConfigureAwait(false);
 140            _created = true;
 141        }
 142        finally
 143        {
 144            _ensureGate.Release();
 145        }
 146    }
 147
 148    /// <summary>
 149    /// Publishes a queue row. The caller supplies the id so a retried publish is idempotent
 150    /// (insert-if-absent) rather than inserting a duplicate job.
 151    /// </summary>
 152    public async Task PublishAsync(
 153        Guid id,
 154        string queue,
 155        string payload,
 156        IReadOnlyDictionary<string, string>? headers,
 157        CancellationToken cancellationToken)
 158    {
 159        await InsertAsync(id, queue, payload, headers, deadLetterReason: null, notify: true, cancellationToken).Configur
 160        await PruneDeadLettersIfDueAsync(cancellationToken).ConfigureAwait(false);
 161    }
 162
 163    public async Task<SqlServerTransportDelivery?> TryClaimAsync(string queue, TimeSpan lockTimeout, CancellationToken c
 164    {
 165        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 166        var lockId = Guid.NewGuid();
 167
 168        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 169        await using var command = connection.CreateCommand();
 170        // READPAST skips rows other subscribers hold UPDLOCK on — SQL Server's equivalent of
 171        // PostgreSQL's FOR UPDATE SKIP LOCKED — so competing consumers never block on each other.
 172        command.CommandText =
 173            $"""
 174            WITH next AS (
 175                SELECT TOP (1) id, queue, payload_json, headers_json, attempts, locked_until, lock_id
 176                FROM {MessageTable} WITH (UPDLOCK, ROWLOCK, READPAST)
 177                WHERE queue = @queue
 178                  AND available_at <= SYSUTCDATETIME()
 179                  AND (locked_until IS NULL OR locked_until <= SYSUTCDATETIME())
 180                ORDER BY created_at
 181            )
 182            UPDATE next
 183            SET attempts = attempts + 1,
 184                locked_until = {AddMilliseconds("@lock_timeout_ms")},
 185                lock_id = @lock_id
 186            OUTPUT inserted.id, inserted.queue, inserted.payload_json, inserted.headers_json, inserted.attempts;
 187            """;
 188        command.Parameters.AddWithValue("@queue", queue);
 189        command.Parameters.AddWithValue("@lock_timeout_ms", (long)lockTimeout.TotalMilliseconds);
 190        command.Parameters.AddWithValue("@lock_id", lockId);
 191
 192        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 193        if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 194            return null;
 195
 196        var id = reader.GetGuid(0);
 197        var payload = reader.GetString(2);
 198        var headerJson = reader.GetString(3);
 199        var attempt = reader.GetInt32(4);
 200        var headers = DeserializeHeaders(headerJson);
 201
 202        return new SqlServerTransportDelivery(
 203            id,
 204            reader.GetString(1),
 205            payload,
 206            headers,
 207            attempt,
 208            () => AckAsync(id, lockId),
 209            delay => NakAsync(id, lockId, delay),
 210            (exception, deleteOriginal, token) => DeadLetterAsync(id, lockId, queue, payload, headers, exception, delete
 211            () => RenewLeaseAsync(id, lockId, lockTimeout));
 212    }
 213
 214    public async IAsyncEnumerable<SqlServerTransportDelivery> ClaimBatchAsync(
 215        string queue,
 216        int batchSize,
 217        TimeSpan lockTimeout,
 218        [EnumeratorCancellation] CancellationToken cancellationToken)
 219    {
 220        for (var i = 0; i < batchSize; i++)
 221        {
 222            var delivery = await TryClaimAsync(queue, lockTimeout, cancellationToken).ConfigureAwait(false);
 223            if (delivery is null)
 224                yield break;
 225            yield return delivery;
 226        }
 227    }
 228
 229    private async Task InsertAsync(
 230        Guid id,
 231        string queue,
 232        string payload,
 233        IReadOnlyDictionary<string, string>? headers,
 234        string? deadLetterReason,
 235        bool notify,
 236        CancellationToken cancellationToken)
 237    {
 238        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 239        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 240        await using var command = connection.CreateCommand();
 241        // Insert-if-absent keeps a retried publish idempotent. The UPDLOCK/HOLDLOCK hints make the
 242        // existence check and the insert atomic; a concurrent same-id insert that still slips through
 243        // surfaces as a duplicate-key error, which is treated as success below.
 244        command.CommandText =
 245            $"""
 246            INSERT INTO {MessageTable} (id, queue, payload_json, headers_json, dead_letter_reason)
 247            SELECT @id, @queue, @payload_json, @headers_json, @dead_letter_reason
 248            WHERE NOT EXISTS (SELECT 1 FROM {MessageTable} WITH (UPDLOCK, HOLDLOCK) WHERE id = @id);
 249            """;
 250        command.Parameters.AddWithValue("@id", id);
 251        command.Parameters.AddWithValue("@queue", queue);
 252        command.Parameters.AddWithValue("@payload_json", payload);
 253        command.Parameters.AddWithValue("@headers_json", AsyncResponseJson.Serialize(headers ?? EmptyHeaders));
 254        command.Parameters.AddWithValue("@dead_letter_reason", (object?)deadLetterReason ?? DBNull.Value);
 255
 256        try
 257        {
 258            await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 259        }
 260        catch (SqlException ex) when (ex.Number is PrimaryKeyViolation or UniqueIndexViolation)
 261        {
 262        }
 263
 264        if (notify)
 265            MessagePublished?.Invoke(queue);
 266    }
 267
 268    private async ValueTask AckAsync(Guid id, Guid lockId)
 269    {
 270        await using var connection = await OpenConnectionAsync(CancellationToken.None).ConfigureAwait(false);
 271        await using var command = connection.CreateCommand();
 272        command.CommandText = $"DELETE FROM {MessageTable} WHERE id = @id AND lock_id = @lock_id;";
 273        command.Parameters.AddWithValue("@id", id);
 274        command.Parameters.AddWithValue("@lock_id", lockId);
 275        await command.ExecuteNonQueryAsync(CancellationToken.None).ConfigureAwait(false);
 276    }
 277
 278    private async ValueTask<bool> RenewLeaseAsync(Guid id, Guid lockId, TimeSpan lockTimeout)
 279    {
 280        await using var connection = await OpenConnectionAsync(CancellationToken.None).ConfigureAwait(false);
 281        await using var command = connection.CreateCommand();
 282        command.CommandText =
 283            $"""
 284            UPDATE {MessageTable}
 285            SET locked_until = {AddMilliseconds("@lock_timeout_ms")}
 286            WHERE id = @id AND lock_id = @lock_id;
 287            """;
 288        command.Parameters.AddWithValue("@id", id);
 289        command.Parameters.AddWithValue("@lock_id", lockId);
 290        command.Parameters.AddWithValue("@lock_timeout_ms", (long)lockTimeout.TotalMilliseconds);
 291        return await command.ExecuteNonQueryAsync(CancellationToken.None).ConfigureAwait(false) > 0;
 292    }
 293
 294    private async ValueTask NakAsync(Guid id, Guid lockId, TimeSpan delay)
 295    {
 296        await using var connection = await OpenConnectionAsync(CancellationToken.None).ConfigureAwait(false);
 297        await using var command = connection.CreateCommand();
 298        command.CommandText =
 299            $"""
 300            UPDATE {MessageTable}
 301            SET available_at = {AddMilliseconds("@delay_ms")},
 302                locked_until = NULL,
 303                lock_id = NULL
 304            WHERE id = @id AND lock_id = @lock_id;
 305            """;
 306        command.Parameters.AddWithValue("@id", id);
 307        command.Parameters.AddWithValue("@lock_id", lockId);
 308        command.Parameters.AddWithValue("@delay_ms", (long)delay.TotalMilliseconds);
 309        await command.ExecuteNonQueryAsync(CancellationToken.None).ConfigureAwait(false);
 310        MessagePublished?.Invoke(null);
 311    }
 312
 313    private async ValueTask<bool> DeadLetterAsync(
 314        Guid id,
 315        Guid lockId,
 316        string sourceQueue,
 317        string payload,
 318        IReadOnlyDictionary<string, string> headers,
 319        Exception exception,
 320        bool deleteOriginal,
 321        CancellationToken cancellationToken)
 322    {
 323        if (!_options.DeadLetterEnabled)
 324        {
 325            if (deleteOriginal)
 326                await AckAsync(id, lockId).ConfigureAwait(false);
 327            return true;
 328        }
 329
 330        var deadHeaders = new Dictionary<string, string>(headers, StringComparer.OrdinalIgnoreCase)
 331        {
 332            ["AR-DeadLetter-Reason"] = Sanitize(exception.Message),
 333            ["AR-DeadLetter-Source-Queue"] = sourceQueue
 334        };
 335
 336        try
 337        {
 338            if (!deleteOriginal)
 339            {
 340                await InsertAsync(Guid.NewGuid(), _options.DeadLetterQueue, payload, deadHeaders, exception.Message, not
 341                return true;
 342            }
 343
 344            // The DLQ insert and the original-row delete must commit atomically: split across two
 345            // connections, a crash between them leaves the original row to be redelivered and
 346            // dead-lettered again, duplicating the DLQ entry.
 347            await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 348            await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 349            await using var transaction = (SqlTransaction)await connection.BeginTransactionAsync(cancellationToken).Conf
 350            await using var command = connection.CreateCommand();
 351            command.Transaction = transaction;
 352            command.CommandText =
 353                $"""
 354                INSERT INTO {MessageTable} (id, queue, payload_json, headers_json, dead_letter_reason)
 355                VALUES (@id, @queue, @payload_json, @headers_json, @dead_letter_reason);
 356                DELETE FROM {MessageTable} WHERE id = @source_id AND lock_id = @lock_id;
 357                """;
 358            command.Parameters.AddWithValue("@id", Guid.NewGuid());
 359            command.Parameters.AddWithValue("@queue", _options.DeadLetterQueue);
 360            command.Parameters.AddWithValue("@payload_json", payload);
 361            command.Parameters.AddWithValue("@headers_json", AsyncResponseJson.Serialize(deadHeaders));
 362            command.Parameters.AddWithValue("@dead_letter_reason", exception.Message);
 363            command.Parameters.AddWithValue("@source_id", id);
 364            command.Parameters.AddWithValue("@lock_id", lockId);
 365            await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 366            await transaction.CommitAsync(cancellationToken).ConfigureAwait(false);
 367            return true;
 368        }
 369        catch (Exception ex)
 370        {
 371            // Callers decide the redelivery consequence from the false return; log the cause here so
 372            // a failing dead-letter write is never silent.
 373            _logger?.LogError(
 374                ex,
 375                "Failed to write SQL Server dead-letter row for message {MessageId} from queue {SourceQueue}.",
 376                id,
 377                sourceQueue);
 378            return false;
 379        }
 380    }
 381
 382    /// <summary>
 383    /// Opportunistically deletes dead-letter rows older than the configured retention. No-op unless
 384    /// <see cref="SqlServerAsyncResponseTransportOptions.DeadLetterRetention"/> is set, and throttled
 385    /// so the DELETE runs at most once per minute regardless of publish rate.
 386    /// </summary>
 387    private async Task PruneDeadLettersIfDueAsync(CancellationToken cancellationToken)
 388    {
 389        if (_options.DeadLetterRetention is not { } retention || !ShouldPruneDeadLetters())
 390            return;
 391
 392        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 393        await using var command = connection.CreateCommand();
 394        command.CommandText = $"DELETE FROM {MessageTable} WHERE queue = @queue AND created_at < {AddMilliseconds("@nega
 395        command.Parameters.AddWithValue("@queue", _options.DeadLetterQueue);
 396        command.Parameters.AddWithValue("@negative_retention_ms", -(long)retention.TotalMilliseconds);
 397        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 398    }
 399
 400    private bool ShouldPruneDeadLetters()
 401    {
 402        var now = DateTime.UtcNow.Ticks;
 403        var last = Interlocked.Read(ref _lastDeadLetterPruneTicks);
 404        return now - last >= DeadLetterPruneThrottle.Ticks
 405            && Interlocked.CompareExchange(ref _lastDeadLetterPruneTicks, now, last) == last;
 406    }
 407
 408    private async Task<SqlConnection> OpenConnectionAsync(CancellationToken cancellationToken)
 409    {
 410        var connection = new SqlConnection(_connectionString);
 411        try
 412        {
 413            await connection.OpenAsync(cancellationToken).ConfigureAwait(false);
 414            return connection;
 415        }
 416        catch
 417        {
 418            await connection.DisposeAsync().ConfigureAwait(false);
 419            throw;
 420        }
 421    }
 422
 423    private static IReadOnlyDictionary<string, string> DeserializeHeaders(string json)
 424    {
 425        var parsed = AsyncResponseJson.Deserialize<Dictionary<string, string>>(json);
 426        return parsed is null
 427            ? EmptyHeaders
 428            : new Dictionary<string, string>(parsed, StringComparer.OrdinalIgnoreCase);
 429    }
 430
 431    private static string Sanitize(string value) => value.Replace('\r', ' ').Replace('\n', ' ');
 432
 433    /// <summary>
 434    /// SQL expression adding a millisecond bigint parameter to the database clock. DATEADD only takes
 435    /// int arguments, so the value is split into whole seconds and a sub-second remainder — intervals
 436    /// (lock timeouts, redelivery delays, retentions) stay on the database clock, immune to app-side
 437    /// clock skew, without overflowing on long spans.
 438    /// </summary>
 439    internal static string AddMilliseconds(string parameterName)
 440        => $"DATEADD(SECOND, CAST({parameterName} / 1000 AS int), DATEADD(MILLISECOND, CAST({parameterName} % 1000 AS in
 441
 442    /// <summary>
 443    /// Stable application-lock resource for serializing schema creation. Must be byte-for-byte
 444    /// identical to the channel store's resource so that, for a shared schema, the channel and
 445    /// transport take the same lock and never race each other on CREATE SCHEMA.
 446    /// </summary>
 447    internal static string SchemaLockResource(string schemaName)
 448        => $"asyncresponse:ddl:{schemaName}";
 449
 450    private static string Quote(string identifier) => "[" + identifier + "]";
 451
 452    private static string IndexName(string table, string suffix)
 453    {
 454        var name = $"{table}_{suffix}_idx";
 455        return name.Length <= 128 ? name : name[..128];
 456    }
 457
 458    private static readonly TimeSpan DeadLetterPruneThrottle = TimeSpan.FromMinutes(1);
 459
 460    private static readonly IReadOnlyDictionary<string, string> EmptyHeaders =
 461        new Dictionary<string, string>(0, StringComparer.OrdinalIgnoreCase);
 462}
Generated by: ReportGenerator 5.5.10.0
08/03/2026 - 08:59:44
GitHub | reportgenerator.io

Methods/Properties

.ctor(System.Guid, string, string, System.Collections.Generic.IReadOnlyDictionary<string, string>, int, System.Func<System.Threading.Tasks.ValueTask>, System.Func<System.TimeSpan, System.Threading.Tasks.ValueTask>, System.Func<System.Exception, bool, System.Threading.CancellationToken, System.Threading.Tasks.ValueTask<bool>>, System.Func<System.Threading.Tasks.ValueTask<bool>>)