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

Information
Class: AsyncResponse.Channels.SqlServer.SqlServerChannelSql
Assembly: AsyncResponse.Channels.SqlServer
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/Channels/AsyncResponse.Channels.SqlServer/SqlServerChannelSql.cs
Line coverage
99%
Covered lines: 410
Uncovered lines: 1
Coverable lines: 411
Total lines: 691
Line coverage: 99.7%
Branch coverage
98%
Covered branches: 83
Total branches: 84
Branch coverage: 98.8%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%11100%
EnsureCreatedAsync()100%66100%
SaveRecoveryStateAsync()100%11100%
LoadRecoveryStatesAsync()100%22100%
DeleteRecoveryStateAsync()100%11100%
ScanRecoveryStateJsonAsync()100%22100%
InsertMessageAsync(...)100%22100%
InsertMessageOnceAsync()100%6696.88%
LoadMessagesAsync()100%88100%
TryClaimForDeliveryAsync()100%22100%
TryClaimForRecoveryAsync()100%22100%
GetServerTimeUtcAsync()100%22100%
IsMessageAcknowledgedAsync()100%22100%
UpsertSubscriberAsync()100%22100%
HeartbeatSubscribersAsync()100%66100%
DeleteSubscriberAsync()100%11100%
CountActiveSubscribersAsync()100%44100%
PruneExpiredRecoveryAsync()100%44100%
PruneExpiredMessagesAsync()100%11100%
PruneExpiredSubscribersAsync()100%44100%
OpenConnectionAsync()50%22100%
ValidateIdentifier(...)100%44100%
IsIdentifier(...)100%1212100%
Quote(...)100%11100%
IndexName(...)100%22100%
AddMilliseconds(...)100%11100%
IsTransient(...)100%66100%
SchemaLockResource(...)100%11100%
ShouldPrune(...)100%44100%

File(s)

/home/runner/work/AsyncResponse/AsyncResponse/src/Channels/AsyncResponse.Channels.SqlServer/SqlServerChannelSql.cs

#LineLine coverage
 1using Microsoft.Data.SqlClient;
 2using System.Data;
 3
 4namespace AsyncResponse.Channels.SqlServer;
 5
 6internal readonly record struct SqlServerChannelMessage(
 7    Guid Id,
 8    string CorrelationId,
 9    string EnvelopeJson,
 10    DateTimeOffset CreatedAtUtc,
 11    DateTimeOffset? AckedAtUtc = null);
 12
 13/// <summary>SQL helper for the SQL Server channel tables.</summary>
 14internal sealed class SqlServerChannelSql
 15{
 16    // SQL Server duplicate-key error numbers: 2627 = PRIMARY KEY/UNIQUE constraint violation,
 17    // 2601 = unique index violation. Retried idempotent inserts treat them as success.
 18    private const int PrimaryKeyViolation = 2627;
 19    private const int UniqueIndexViolation = 2601;
 20
 21    private readonly string _connectionString;
 22    private readonly SqlServerAsyncResponseChannelOptions _options;
 323    private readonly SemaphoreSlim _ensureGate = new(1, 1);
 24    private bool _created;
 25    private long _lastRecoveryPruneTicks;
 26    private long _lastMessagePruneTicks;
 27    private long _lastSubscriberPruneTicks;
 28
 329    public SqlServerChannelSql(Microsoft.Extensions.Options.IOptions<SqlServerAsyncResponseChannelOptions> options)
 30    {
 331        _options = options.Value;
 332        _options.Validate();
 333        _connectionString = _options.ConnectionString!;
 34
 335        Schema = Quote(_options.SchemaName);
 336        RecoveryTable = $"{Schema}.{Quote(_options.RecoveryStateTable)}";
 337        MessageTable = $"{Schema}.{Quote(_options.MessageTable)}";
 338        SubscriberTable = $"{Schema}.{Quote(_options.SubscriberTable)}";
 339    }
 40
 41    public string Schema { get; }
 42    public string RecoveryTable { get; }
 43    public string MessageTable { get; }
 44    public string SubscriberTable { get; }
 45
 46    public async Task EnsureCreatedAsync(CancellationToken cancellationToken = default)
 47    {
 348        if (_created || !_options.AutoCreateSchema)
 349            return;
 50
 351        await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 52        try
 53        {
 354            if (_created)
 155                return;
 56
 357            await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 158            await using var transaction = (SqlTransaction)await connection.BeginTransactionAsync(cancellationToken).Conf
 59
 60            // Serialize schema creation across processes. The IF-NOT-EXISTS guards are not atomic
 61            // against a concurrent create of the same object: two instances starting together both
 62            // pass the existence check and collide on the catalog (error 2714/2627). A
 63            // transaction-scoped application lock (keyed by schema, shared with the transport store)
 64            // lets one instance build the schema while the rest wait and then find it already present.
 165            await using (var lockCommand = connection.CreateCommand())
 66            {
 167                lockCommand.Transaction = transaction;
 168                lockCommand.CommandText =
 169                    """
 170                    DECLARE @lock_result int;
 171                    EXEC @lock_result = sp_getapplock
 172                        @Resource = @lock_resource,
 173                        @LockMode = 'Exclusive',
 174                        @LockOwner = 'Transaction',
 175                        @LockTimeout = 60000;
 176                    IF @lock_result < 0
 177                        THROW 51000, N'Failed to acquire the AsyncResponse DDL application lock.', 1;
 178                    """;
 179                lockCommand.Parameters.AddWithValue("@lock_resource", SchemaLockResource(_options.SchemaName));
 180                await lockCommand.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 81            }
 82
 183            await using var command = connection.CreateCommand();
 184            command.Transaction = transaction;
 185            command.CommandText =
 186                $"""
 187                IF SCHEMA_ID(N'{_options.SchemaName}') IS NULL
 188                    EXEC(N'CREATE SCHEMA {Schema}');
 189
 190                IF OBJECT_ID(N'{RecoveryTable}', N'U') IS NULL
 191                CREATE TABLE {RecoveryTable} (
 192                    correlation_id nvarchar(400) NOT NULL,
 193                    registration_id uniqueidentifier NOT NULL,
 194                    state_json nvarchar(max) NOT NULL,
 195                    expires_at datetime2 NOT NULL,
 196                    registered_at datetime2 NOT NULL DEFAULT SYSUTCDATETIME(),
 197                    PRIMARY KEY (correlation_id, registration_id)
 198                );
 199                IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name = N'{IndexName(_options.RecoveryStateTable, "expires
 1100                    CREATE INDEX {Quote(IndexName(_options.RecoveryStateTable, "expires"))}
 1101                        ON {RecoveryTable} (expires_at);
 1102
 1103                IF OBJECT_ID(N'{MessageTable}', N'U') IS NULL
 1104                CREATE TABLE {MessageTable} (
 1105                    id uniqueidentifier NOT NULL PRIMARY KEY NONCLUSTERED,
 1106                    correlation_id nvarchar(400) NOT NULL,
 1107                    envelope_json nvarchar(max) NOT NULL,
 1108                    created_at datetime2 NOT NULL DEFAULT SYSUTCDATETIME(),
 1109                    expires_at datetime2 NOT NULL,
 1110                    acked_at datetime2 NULL,
 1111                    recovery_claimed bit NOT NULL DEFAULT 0
 1112                );
 1113                IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name = N'{IndexName(_options.MessageTable, "correlation_c
 1114                    CREATE INDEX {Quote(IndexName(_options.MessageTable, "correlation_created"))}
 1115                        ON {MessageTable} (correlation_id, created_at);
 1116                IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name = N'{IndexName(_options.MessageTable, "expires")}' A
 1117                    CREATE INDEX {Quote(IndexName(_options.MessageTable, "expires"))}
 1118                        ON {MessageTable} (expires_at);
 1119
 1120                IF OBJECT_ID(N'{SubscriberTable}', N'U') IS NULL
 1121                CREATE TABLE {SubscriberTable} (
 1122                    correlation_id nvarchar(400) NOT NULL,
 1123                    registration_id uniqueidentifier NOT NULL,
 1124                    instance_id nvarchar(200) NOT NULL,
 1125                    expires_at datetime2 NOT NULL,
 1126                    PRIMARY KEY (correlation_id, registration_id)
 1127                );
 1128                IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name = N'{IndexName(_options.SubscriberTable, "expires")}
 1129                    CREATE INDEX {Quote(IndexName(_options.SubscriberTable, "expires"))}
 1130                        ON {SubscriberTable} (expires_at);
 1131                """;
 1132            await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 1133            await transaction.CommitAsync(cancellationToken).ConfigureAwait(false);
 1134            _created = true;
 1135        }
 136        finally
 137        {
 3138            _ensureGate.Release();
 139        }
 3140    }
 141
 142    public async Task SaveRecoveryStateAsync(string correlationId, RecoveryState state, TimeSpan ttl, CancellationToken 
 143    {
 3144        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 3145        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1146        await using var command = connection.CreateCommand();
 147        // MERGE WITH (HOLDLOCK) makes the match check and insert atomic — the SQL Server equivalent
 148        // of PostgreSQL's INSERT ... ON CONFLICT DO UPDATE for the (correlation_id, registration_id) key.
 1149        command.CommandText =
 1150            $"""
 1151            MERGE {RecoveryTable} WITH (HOLDLOCK) AS target
 1152            USING (SELECT @correlation_id AS correlation_id, @registration_id AS registration_id) AS source
 1153                ON target.correlation_id = source.correlation_id AND target.registration_id = source.registration_id
 1154            WHEN MATCHED THEN
 1155                UPDATE SET state_json = @state_json,
 1156                           expires_at = {AddMilliseconds("@ttl_ms")},
 1157                           registered_at = SYSUTCDATETIME()
 1158            WHEN NOT MATCHED THEN
 1159                INSERT (correlation_id, registration_id, state_json, expires_at, registered_at)
 1160                VALUES (@correlation_id, @registration_id, @state_json, {AddMilliseconds("@ttl_ms")}, SYSUTCDATETIME());
 1161            """;
 1162        command.Parameters.AddWithValue("@correlation_id", correlationId);
 1163        command.Parameters.AddWithValue("@registration_id", state.RegistrationId);
 1164        command.Parameters.AddWithValue("@state_json", AsyncResponseJson.Serialize(state));
 1165        command.Parameters.AddWithValue("@ttl_ms", (long)ttl.TotalMilliseconds);
 1166        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 1167    }
 168
 169    public async Task<IReadOnlyList<string>> LoadRecoveryStatesAsync(string correlationId, CancellationToken cancellatio
 170    {
 1171        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 1172        if (ShouldPrune(ref _lastRecoveryPruneTicks))
 1173            await PruneExpiredRecoveryAsync(correlationId, cancellationToken).ConfigureAwait(false);
 174
 1175        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1176        await using var command = connection.CreateCommand();
 1177        command.CommandText =
 1178            $"""
 1179            SELECT state_json
 1180            FROM {RecoveryTable}
 1181            WHERE correlation_id = @correlation_id AND expires_at > SYSUTCDATETIME()
 1182            ORDER BY registered_at;
 1183            """;
 1184        command.Parameters.AddWithValue("@correlation_id", correlationId);
 185
 1186        var states = new List<string>();
 1187        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 1188        while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 1189            states.Add(reader.GetString(0));
 1190        return states;
 1191    }
 192
 193    public async Task<bool> DeleteRecoveryStateAsync(string correlationId, Guid registrationId, CancellationToken cancel
 194    {
 1195        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 1196        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1197        await using var command = connection.CreateCommand();
 1198        command.CommandText = $"DELETE FROM {RecoveryTable} WHERE correlation_id = @correlation_id AND registration_id =
 1199        command.Parameters.AddWithValue("@correlation_id", correlationId);
 1200        command.Parameters.AddWithValue("@registration_id", registrationId);
 1201        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 1202    }
 203
 204    public async IAsyncEnumerable<string> ScanRecoveryStateJsonAsync([System.Runtime.CompilerServices.EnumeratorCancella
 205    {
 1206        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 1207        await PruneExpiredRecoveryAsync(null, cancellationToken).ConfigureAwait(false);
 208
 1209        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1210        await using var command = connection.CreateCommand();
 1211        command.CommandText =
 1212            $"""
 1213            SELECT state_json
 1214            FROM {RecoveryTable}
 1215            WHERE expires_at > SYSUTCDATETIME()
 1216            ORDER BY registered_at;
 1217            """;
 1218        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 1219        while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 1220            yield return reader.GetString(0);
 1221    }
 222
 223    /// <summary>
 224    /// Inserts a response envelope row. The caller supplies the message id so the insert is
 225    /// idempotent under retry — a duplicate insert (lost WHERE NOT EXISTS race or an outer retry)
 226    /// is treated as success, so a retried publish never duplicates a response. Returns the row's
 227    /// server-stamped <c>created_at</c> (the original row's on a duplicate) so the same-process
 228    /// fast path compares against subscription watermarks on the server clock rather than the
 229    /// app clock.
 230    /// </summary>
 231    public Task<DateTimeOffset> InsertMessageAsync(Guid id, string correlationId, string envelopeJson, TimeSpan retentio
 1232        => AsyncResponseRetry.ExecuteAsync(
 1233            token => InsertMessageOnceAsync(id, correlationId, envelopeJson, retention, token),
 1234            IsTransient,
 1235            _options.PublishMaxAttempts,
 1236            _options.PublishRetryBaseDelay,
 1237            _options.PublishRetryMaxDelay,
 1238            cancellationToken);
 239
 240    private async Task<DateTimeOffset> InsertMessageOnceAsync(Guid id, string correlationId, string envelopeJson, TimeSp
 241    {
 1242        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 1243        if (ShouldPrune(ref _lastMessagePruneTicks))
 1244            await PruneExpiredMessagesAsync(cancellationToken).ConfigureAwait(false);
 245
 1246        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1247        await using var command = connection.CreateCommand();
 1248        command.CommandText =
 1249            $"""
 1250            INSERT INTO {MessageTable} (id, correlation_id, envelope_json, expires_at)
 1251            OUTPUT inserted.created_at
 1252            SELECT @id, @correlation_id, @envelope_json, {AddMilliseconds("@retention_ms")}
 1253            WHERE NOT EXISTS (SELECT 1 FROM {MessageTable} WITH (UPDLOCK, HOLDLOCK) WHERE id = @id);
 1254            """;
 1255        command.Parameters.AddWithValue("@id", id);
 1256        command.Parameters.AddWithValue("@correlation_id", correlationId);
 1257        command.Parameters.AddWithValue("@envelope_json", envelopeJson);
 1258        command.Parameters.AddWithValue("@retention_ms", (long)retention.TotalMilliseconds);
 259
 1260        object? createdAt = null;
 261        try
 262        {
 1263            createdAt = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
 1264        }
 0265        catch (SqlException ex) when (ex.Number is PrimaryKeyViolation or UniqueIndexViolation)
 266        {
 1267        }
 268
 1269        if (createdAt is DateTime insertedCreatedAt)
 1270            return new DateTimeOffset(insertedCreatedAt, TimeSpan.Zero);
 271
 272        // Duplicate insert (WHERE NOT EXISTS suppressed it, or the key-violation race lost):
 273        // return the original row's server-stamped created_at. Unlike PostgreSQL's single-statement
 274        // CTE, this fallback is already a SEPARATE statement, so a concurrent same-id publish is
 275        // resolved here deterministically: the HOLDLOCK range lock on the first statement
 276        // serializes against the competing insert, and this second statement reads its own fresh
 277        // snapshot/locks and sees the committed row.
 1278        await using var lookup = connection.CreateCommand();
 1279        lookup.CommandText = $"SELECT created_at FROM {MessageTable} WHERE id = @id;";
 1280        lookup.Parameters.AddWithValue("@id", id);
 1281        var existing = await lookup.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
 282
 283        // A missing row means the idempotent duplicate's original is already gone (pruned
 284        // mid-publish): the message is not persisted, so reporting success with a fabricated
 285        // app-clock timestamp would both lie about persistence and feed a client clock into the
 286        // server-clock watermark. Fail instead, so the publisher's error handling runs.
 1287        return existing is DateTime existingCreatedAt
 1288            ? new DateTimeOffset(existingCreatedAt, TimeSpan.Zero)
 1289            : throw new InvalidOperationException(
 1290                $"SQL Server response insert for message {id} found no row after a duplicate: the original no longer exi
 1291    }
 292
 293    public async Task<IReadOnlyList<SqlServerChannelMessage>> LoadMessagesAsync(
 294        string correlationId,
 295        DateTimeOffset sinceUtc,
 296        int batchSize,
 297        DateTimeOffset? afterCreatedAtUtc,
 298        Guid? afterId,
 299        CancellationToken cancellationToken)
 300    {
 3301        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 3302        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1303        await using var command = connection.CreateCommand();
 1304        command.CommandText =
 1305            $"""
 1306            SELECT id, correlation_id, envelope_json, created_at, acked_at
 1307            FROM {MessageTable}
 1308            WHERE correlation_id = @correlation_id
 1309              AND created_at >= @since
 1310              AND expires_at > SYSUTCDATETIME()
 1311              {(afterCreatedAtUtc is null ? "" : "AND (created_at > @after_created_at OR (created_at = @after_created_at
 1312            ORDER BY created_at, id
 1313            OFFSET 0 ROWS FETCH NEXT @limit ROWS ONLY;
 1314            """;
 1315        command.Parameters.AddWithValue("@correlation_id", correlationId);
 1316        var sinceParameter = command.Parameters.Add("@since", SqlDbType.DateTime2);
 1317        sinceParameter.Scale = 7;
 1318        sinceParameter.Value = sinceUtc.UtcDateTime;
 1319        command.Parameters.AddWithValue("@limit", batchSize);
 1320        if (afterCreatedAtUtc is not null)
 321        {
 1322            var cursorParameter = command.Parameters.Add("@after_created_at", SqlDbType.DateTime2);
 1323            cursorParameter.Scale = 7;
 1324            cursorParameter.Value = afterCreatedAtUtc.Value.UtcDateTime;
 1325            command.Parameters.AddWithValue("@after_id", afterId ?? throw new ArgumentNullException(nameof(afterId)));
 326        }
 327
 1328        var messages = new List<SqlServerChannelMessage>(batchSize);
 1329        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 1330        while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 1331            messages.Add(new SqlServerChannelMessage(
 1332                reader.GetGuid(0),
 1333                reader.GetString(1),
 1334                reader.GetString(2),
 1335                new DateTimeOffset(reader.GetDateTime(3), TimeSpan.Zero),
 1336                reader.IsDBNull(4) ? null : new DateTimeOffset(reader.GetDateTime(4), TimeSpan.Zero)));
 1337        return messages;
 1338    }
 339
 340    /// <summary>
 341    /// Atomically claims a message for live delivery: sets <c>acked_at</c> unless the publisher has
 342    /// already routed it to the lost-subscriber path (<c>recovery_claimed</c>). Returns <c>false</c>
 343    /// when recovery owns the message, so a slow-but-live waiter does not deliver a response the
 344    /// recovery callback already handled. Multiple processes may each win this claim, preserving
 345    /// cross-process fan-out, because it gates only on <c>recovery_claimed</c>, not on <c>acked_at</c>.
 346    /// </summary>
 347    public async Task<bool> TryClaimForDeliveryAsync(Guid messageId, CancellationToken cancellationToken)
 348    {
 3349        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 3350        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1351        await using var command = connection.CreateCommand();
 1352        command.CommandText =
 1353            $"""
 1354            UPDATE {MessageTable}
 1355            SET acked_at = COALESCE(acked_at, SYSUTCDATETIME())
 1356            OUTPUT inserted.id
 1357            WHERE id = @id AND recovery_claimed = 0 AND expires_at > SYSUTCDATETIME();
 1358            """;
 1359        command.Parameters.AddWithValue("@id", messageId);
 1360        var result = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
 1361        return result is not null and not DBNull;
 1362    }
 363
 364    /// <summary>
 365    /// Atomically claims a message for the lost-subscriber path: sets <c>recovery_claimed</c> only
 366    /// while no waiter has delivered (<c>acked_at IS NULL</c>). Returns <c>true</c> when recovery wins;
 367    /// <c>false</c> means a live waiter already took the message, so the publisher must not also fire
 368    /// the recovery callback. Row-level locking serializes this against <see cref="TryClaimForDeliveryAsync"/>.
 369    /// </summary>
 370    public async Task<bool> TryClaimForRecoveryAsync(Guid messageId, CancellationToken cancellationToken)
 371    {
 1372        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 1373        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1374        await using var command = connection.CreateCommand();
 1375        command.CommandText =
 1376            $"""
 1377            UPDATE {MessageTable}
 1378            SET recovery_claimed = 1
 1379            OUTPUT inserted.id
 1380            WHERE id = @id AND acked_at IS NULL;
 1381            """;
 1382        command.Parameters.AddWithValue("@id", messageId);
 1383        var result = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
 1384        return result is not null and not DBNull;
 1385    }
 386
 387    /// <summary>Returns the database server's current UTC time, used as a clock-safe delivery watermark.</summary>
 388    public async Task<DateTimeOffset> GetServerTimeUtcAsync(CancellationToken cancellationToken)
 389    {
 3390        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 3391        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1392        await using var command = connection.CreateCommand();
 1393        command.CommandText = "SELECT SYSUTCDATETIME();";
 1394        var result = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
 1395        return result switch
 1396        {
 1397            DateTimeOffset dto => dto.ToUniversalTime(),
 1398            DateTime dt => new DateTimeOffset(DateTime.SpecifyKind(dt, DateTimeKind.Utc), TimeSpan.Zero),
 1399            _ => DateTimeOffset.UtcNow
 1400        };
 1401    }
 402
 403    public async Task<bool> IsMessageAcknowledgedAsync(Guid messageId, CancellationToken cancellationToken)
 404    {
 3405        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 3406        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1407        await using var command = connection.CreateCommand();
 1408        command.CommandText =
 1409            $"""
 1410            SELECT CAST(CASE WHEN acked_at IS NOT NULL THEN 1 ELSE 0 END AS bit)
 1411            FROM {MessageTable}
 1412            WHERE id = @id AND expires_at > SYSUTCDATETIME();
 1413            """;
 1414        command.Parameters.AddWithValue("@id", messageId);
 1415        var result = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
 1416        return result is bool acknowledged && acknowledged;
 1417    }
 418
 419    public async Task UpsertSubscriberAsync(string correlationId, Guid registrationId, string instanceId, TimeSpan ttl, 
 420    {
 1421        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 1422        if (ShouldPrune(ref _lastSubscriberPruneTicks))
 1423            await PruneExpiredSubscribersAsync(correlationId, cancellationToken).ConfigureAwait(false);
 424
 1425        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1426        await using var command = connection.CreateCommand();
 1427        command.CommandText =
 1428            $"""
 1429            MERGE {SubscriberTable} WITH (HOLDLOCK) AS target
 1430            USING (SELECT @correlation_id AS correlation_id, @registration_id AS registration_id) AS source
 1431                ON target.correlation_id = source.correlation_id AND target.registration_id = source.registration_id
 1432            WHEN MATCHED THEN
 1433                UPDATE SET instance_id = @instance_id,
 1434                           expires_at = {AddMilliseconds("@ttl_ms")}
 1435            WHEN NOT MATCHED THEN
 1436                INSERT (correlation_id, registration_id, instance_id, expires_at)
 1437                VALUES (@correlation_id, @registration_id, @instance_id, {AddMilliseconds("@ttl_ms")});
 1438            """;
 1439        command.Parameters.AddWithValue("@correlation_id", correlationId);
 1440        command.Parameters.AddWithValue("@registration_id", registrationId);
 1441        command.Parameters.AddWithValue("@instance_id", instanceId);
 1442        command.Parameters.AddWithValue("@ttl_ms", (long)ttl.TotalMilliseconds);
 1443        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 1444    }
 445
 446    public async Task HeartbeatSubscribersAsync(
 447        string instanceId,
 448        IReadOnlyList<(string CorrelationId, Guid RegistrationId)> registrations,
 449        TimeSpan ttl,
 450        CancellationToken cancellationToken)
 451    {
 3452        if (registrations.Count == 0)
 3453            return;
 454
 3455        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 3456        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 457
 458        // Two parameters per row plus instance/ttl stays under SQL Server's 2100-parameter cap.
 459        const int batchSize = 1000;
 1460        for (var offset = 0; offset < registrations.Count; offset += batchSize)
 461        {
 1462            var count = Math.Min(batchSize, registrations.Count - offset);
 1463            await using var command = connection.CreateCommand();
 1464            var sourceRows = new string[count];
 1465            for (var index = 0; index < count; index++)
 466            {
 1467                var (correlationId, registrationId) = registrations[offset + index];
 1468                sourceRows[index] = $"(@correlation_id_{index}, @registration_id_{index})";
 1469                command.Parameters.AddWithValue($"@correlation_id_{index}", correlationId);
 1470                command.Parameters.AddWithValue($"@registration_id_{index}", registrationId);
 471            }
 472
 473            // MERGE upsert rather than a bare UPDATE, in the same WITH (HOLDLOCK) style as
 474            // UpsertSubscriberAsync: the caller only heartbeats registrations that are live in this
 475            // process, so a missing row means the pruner deleted it (e.g. after a >timeout stall)
 476            // — re-creating it here is what brings the waiter back from "permanently invisible".
 1477            command.CommandText =
 1478                $"""
 1479                MERGE {SubscriberTable} WITH (HOLDLOCK) AS target
 1480                USING (VALUES {string.Join(", ", sourceRows)}) AS source (correlation_id, registration_id)
 1481                    ON target.correlation_id = source.correlation_id AND target.registration_id = source.registration_id
 1482                WHEN MATCHED THEN
 1483                    UPDATE SET instance_id = @instance_id,
 1484                               expires_at = {AddMilliseconds("@ttl_ms")}
 1485                WHEN NOT MATCHED THEN
 1486                    INSERT (correlation_id, registration_id, instance_id, expires_at)
 1487                    VALUES (source.correlation_id, source.registration_id, @instance_id, {AddMilliseconds("@ttl_ms")});
 1488                """;
 1489            command.Parameters.AddWithValue("@instance_id", instanceId);
 1490            command.Parameters.AddWithValue("@ttl_ms", (long)ttl.TotalMilliseconds);
 1491            await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 1492        }
 3493    }
 494
 495    public async Task DeleteSubscriberAsync(string correlationId, Guid registrationId, CancellationToken cancellationTok
 496    {
 3497        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 3498        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1499        await using var command = connection.CreateCommand();
 1500        command.CommandText = $"DELETE FROM {SubscriberTable} WHERE correlation_id = @correlation_id AND registration_id
 1501        command.Parameters.AddWithValue("@correlation_id", correlationId);
 1502        command.Parameters.AddWithValue("@registration_id", registrationId);
 1503        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 1504    }
 505
 506    public async Task<long> CountActiveSubscribersAsync(string correlationId, CancellationToken cancellationToken)
 507    {
 3508        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 3509        if (ShouldPrune(ref _lastSubscriberPruneTicks))
 3510            await PruneExpiredSubscribersAsync(correlationId, cancellationToken).ConfigureAwait(false);
 511
 3512        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1513        await using var command = connection.CreateCommand();
 1514        command.CommandText =
 1515            $"""
 1516            SELECT COUNT_BIG(*)
 1517            FROM {SubscriberTable}
 1518            WHERE correlation_id = @correlation_id AND expires_at > SYSUTCDATETIME();
 1519            """;
 1520        command.Parameters.AddWithValue("@correlation_id", correlationId);
 1521        var result = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
 1522        return result is long count ? count : 0L;
 1523    }
 524
 525    private async Task PruneExpiredRecoveryAsync(string? correlationId, CancellationToken cancellationToken)
 526    {
 3527        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1528        await using var command = connection.CreateCommand();
 1529        command.CommandText = correlationId is null
 1530            ? $"DELETE FROM {RecoveryTable} WHERE expires_at <= SYSUTCDATETIME();"
 1531            : $"DELETE FROM {RecoveryTable} WHERE correlation_id = @correlation_id AND expires_at <= SYSUTCDATETIME();";
 1532        if (correlationId is not null)
 1533            command.Parameters.AddWithValue("@correlation_id", correlationId);
 1534        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 1535    }
 536
 537    private async Task PruneExpiredMessagesAsync(CancellationToken cancellationToken)
 538    {
 3539        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1540        await using var command = connection.CreateCommand();
 1541        command.CommandText = $"DELETE FROM {MessageTable} WHERE expires_at <= SYSUTCDATETIME();";
 1542        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 1543    }
 544
 545    private async Task PruneExpiredSubscribersAsync(string? correlationId, CancellationToken cancellationToken)
 546    {
 3547        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1548        await using var command = connection.CreateCommand();
 1549        command.CommandText = correlationId is null
 1550            ? $"DELETE FROM {SubscriberTable} WHERE expires_at <= SYSUTCDATETIME();"
 1551            : $"DELETE FROM {SubscriberTable} WHERE correlation_id = @correlation_id AND expires_at <= SYSUTCDATETIME();
 1552        if (correlationId is not null)
 1553            command.Parameters.AddWithValue("@correlation_id", correlationId);
 1554        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 1555    }
 556
 557    private async Task<SqlConnection> OpenConnectionAsync(CancellationToken cancellationToken)
 558    {
 3559        var connection = new SqlConnection(_connectionString);
 560        try
 561        {
 3562            await connection.OpenAsync(cancellationToken).ConfigureAwait(false);
 1563            return connection;
 564        }
 3565        catch
 566        {
 3567            await connection.DisposeAsync().ConfigureAwait(false);
 3568            throw;
 569        }
 1570    }
 571
 572    public static void ValidateIdentifier(string? value, string name)
 573    {
 3574        if (string.IsNullOrWhiteSpace(value))
 3575            throw new InvalidOperationException($"{nameof(SqlServerAsyncResponseChannelOptions)}.{name} must be configur
 3576        if (!IsIdentifier(value))
 3577            throw new InvalidOperationException(
 3578                $"{nameof(SqlServerAsyncResponseChannelOptions)}.{name} '{value}' must be a simple SQL Server identifier
 3579    }
 580
 581    private static bool IsIdentifier(string value)
 582    {
 3583        if (value.Length == 0 || !(char.IsAsciiLetter(value[0]) || value[0] == '_'))
 3584            return false;
 585
 3586        foreach (var c in value)
 587        {
 3588            if (!(char.IsAsciiLetterOrDigit(c) || c == '_'))
 3589                return false;
 590        }
 591
 3592        return true;
 593    }
 594
 3595    private static string Quote(string identifier) => "[" + identifier + "]";
 596
 597    private static string IndexName(string table, string suffix)
 598    {
 3599        var name = $"{table}_{suffix}_idx";
 3600        return name.Length <= 128 ? name : name[..128];
 601    }
 602
 603    /// <summary>
 604    /// SQL expression adding a millisecond bigint parameter to the database clock. DATEADD only takes
 605    /// int arguments, so the value is split into whole seconds and a sub-second remainder — TTLs and
 606    /// retentions stay on the database clock, immune to app-side clock skew, without overflowing on
 607    /// long spans such as the 7-day recovery expiry.
 608    /// </summary>
 609    internal static string AddMilliseconds(string parameterName)
 3610        => $"DATEADD(SECOND, CAST({parameterName} / 1000 AS int), DATEADD(MILLISECOND, CAST({parameterName} % 1000 AS in
 611
 612    internal static bool IsTransient(Exception exception)
 3613        => exception is not OperationCanceledException
 3614           && (exception is SqlException sqlException && SqlServerTransientFaults.IsTransient(sqlException)
 3615               || exception is TimeoutException);
 616
 617    /// <summary>
 618    /// Stable application-lock resource for serializing schema creation. It must be deterministic
 619    /// across processes and identical to the transport store's resource for the same schema so both
 620    /// serialize their shared CREATE SCHEMA.
 621    /// </summary>
 622    internal static string SchemaLockResource(string schemaName)
 3623        => $"asyncresponse:ddl:{schemaName}";
 624
 625    /// <summary>
 626    /// Time-gates opportunistic pruning so the housekeeping DELETE runs at most once per
 627    /// <see cref="SqlServerAsyncResponseChannelOptions.PruneInterval"/> instead of on every operation.
 628    /// Read queries already filter on <c>expires_at</c>, so throttling pruning never affects correctness.
 629    /// </summary>
 630    private bool ShouldPrune(ref long lastTicks)
 631    {
 3632        var interval = _options.PruneInterval;
 3633        if (interval <= TimeSpan.Zero)
 3634            return true;
 635
 3636        var now = DateTime.UtcNow.Ticks;
 3637        var last = Interlocked.Read(ref lastTicks);
 3638        return now - last >= interval.Ticks
 3639            && Interlocked.CompareExchange(ref lastTicks, now, last) == last;
 640    }
 641}
 642
 643/// <summary>
 644/// Classifies SQL Server errors worth retrying. <see cref="SqlException"/> exposes no public
 645/// transient flag, so this mirrors the error numbers Microsoft's own retry guidance and the
 646/// SqlClient configurable-retry defaults treat as transient, plus severity ≥ 20 (broken connection).
 647/// </summary>
 648internal static class SqlServerTransientFaults
 649{
 650    private static readonly HashSet<int> TransientErrorNumbers =
 651    [
 652        -2,    // client-side command timeout
 653        20,    // instance does not support encryption
 654        64,    // connection lost during login
 655        121,   // transport semaphore timeout
 656        233,   // no process on the other end of the pipe
 657        997,   // overlapped I/O in progress
 658        1204,  // lock resources exhausted
 659        1205,  // deadlock victim
 660        1222,  // lock request timeout
 661        4060,  // database unavailable
 662        4221,  // readable secondary timeout
 663        10053, // transport-level connection abort
 664        10054, // transport-level connection reset
 665        10060, // network unreachable / connect timeout
 666        10928, // Azure SQL resource limit reached
 667        10929, // Azure SQL minimum guarantee exceeded
 668        40143, // Azure SQL connection failure
 669        40197, // Azure SQL service processing error
 670        40501, // Azure SQL service busy
 671        40540, // Azure SQL service unavailable
 672        40613, // Azure SQL database unavailable
 673        49918, // cannot process request, not enough resources
 674        49919, // cannot process create/update request
 675        49920  // cannot process request, too many operations
 676    ];
 677
 678    public static bool IsTransient(SqlException exception)
 679    {
 680        if (exception.Class >= 20)
 681            return true;
 682
 683        foreach (SqlError error in exception.Errors)
 684        {
 685            if (TransientErrorNumbers.Contains(error.Number))
 686                return true;
 687        }
 688
 689        return TransientErrorNumbers.Contains(exception.Number);
 690    }
 691}
Generated by: ReportGenerator 5.5.10.0
08/03/2026 - 08:59:44
GitHub | reportgenerator.io

Methods/Properties

.ctor(Microsoft.Extensions.Options.IOptions<AsyncResponse.Channels.SqlServer.SqlServerAsyncResponseChannelOptions>)
EnsureCreatedAsync()
SaveRecoveryStateAsync()
LoadRecoveryStatesAsync()
DeleteRecoveryStateAsync()
ScanRecoveryStateJsonAsync()
InsertMessageAsync(System.Guid, string, string, System.TimeSpan, System.Threading.CancellationToken)
InsertMessageOnceAsync()
LoadMessagesAsync()
TryClaimForDeliveryAsync()
TryClaimForRecoveryAsync()
GetServerTimeUtcAsync()
IsMessageAcknowledgedAsync()
UpsertSubscriberAsync()
HeartbeatSubscribersAsync()
DeleteSubscriberAsync()
CountActiveSubscribersAsync()
PruneExpiredRecoveryAsync()
PruneExpiredMessagesAsync()
PruneExpiredSubscribersAsync()
OpenConnectionAsync()
ValidateIdentifier(string, string)
IsIdentifier(string)
Quote(string)
IndexName(string, string)
AddMilliseconds(string)
IsTransient(System.Exception)
SchemaLockResource(string)
ShouldPrune(ref long)