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

Information
Class: AsyncResponse.Channels.SqlServer.SqlServerTransientFaults
Assembly: AsyncResponse.Channels.SqlServer
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/Channels/AsyncResponse.Channels.SqlServer/SqlServerChannelSql.cs
Line coverage
100%
Covered lines: 34
Uncovered lines: 0
Coverable lines: 34
Total lines: 691
Line coverage: 100%
Branch coverage
100%
Covered branches: 6
Total branches: 6
Branch coverage: 100%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.cctor()100%11100%
IsTransient(...)100%66100%

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;
 23    private readonly SemaphoreSlim _ensureGate = new(1, 1);
 24    private bool _created;
 25    private long _lastRecoveryPruneTicks;
 26    private long _lastMessagePruneTicks;
 27    private long _lastSubscriberPruneTicks;
 28
 29    public SqlServerChannelSql(Microsoft.Extensions.Options.IOptions<SqlServerAsyncResponseChannelOptions> options)
 30    {
 31        _options = options.Value;
 32        _options.Validate();
 33        _connectionString = _options.ConnectionString!;
 34
 35        Schema = Quote(_options.SchemaName);
 36        RecoveryTable = $"{Schema}.{Quote(_options.RecoveryStateTable)}";
 37        MessageTable = $"{Schema}.{Quote(_options.MessageTable)}";
 38        SubscriberTable = $"{Schema}.{Quote(_options.SubscriberTable)}";
 39    }
 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    {
 48        if (_created || !_options.AutoCreateSchema)
 49            return;
 50
 51        await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 52        try
 53        {
 54            if (_created)
 55                return;
 56
 57            await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 58            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.
 65            await using (var lockCommand = connection.CreateCommand())
 66            {
 67                lockCommand.Transaction = transaction;
 68                lockCommand.CommandText =
 69                    """
 70                    DECLARE @lock_result int;
 71                    EXEC @lock_result = sp_getapplock
 72                        @Resource = @lock_resource,
 73                        @LockMode = 'Exclusive',
 74                        @LockOwner = 'Transaction',
 75                        @LockTimeout = 60000;
 76                    IF @lock_result < 0
 77                        THROW 51000, N'Failed to acquire the AsyncResponse DDL application lock.', 1;
 78                    """;
 79                lockCommand.Parameters.AddWithValue("@lock_resource", SchemaLockResource(_options.SchemaName));
 80                await lockCommand.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 81            }
 82
 83            await using var command = connection.CreateCommand();
 84            command.Transaction = transaction;
 85            command.CommandText =
 86                $"""
 87                IF SCHEMA_ID(N'{_options.SchemaName}') IS NULL
 88                    EXEC(N'CREATE SCHEMA {Schema}');
 89
 90                IF OBJECT_ID(N'{RecoveryTable}', N'U') IS NULL
 91                CREATE TABLE {RecoveryTable} (
 92                    correlation_id nvarchar(400) NOT NULL,
 93                    registration_id uniqueidentifier NOT NULL,
 94                    state_json nvarchar(max) NOT NULL,
 95                    expires_at datetime2 NOT NULL,
 96                    registered_at datetime2 NOT NULL DEFAULT SYSUTCDATETIME(),
 97                    PRIMARY KEY (correlation_id, registration_id)
 98                );
 99                IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name = N'{IndexName(_options.RecoveryStateTable, "expires
 100                    CREATE INDEX {Quote(IndexName(_options.RecoveryStateTable, "expires"))}
 101                        ON {RecoveryTable} (expires_at);
 102
 103                IF OBJECT_ID(N'{MessageTable}', N'U') IS NULL
 104                CREATE TABLE {MessageTable} (
 105                    id uniqueidentifier NOT NULL PRIMARY KEY NONCLUSTERED,
 106                    correlation_id nvarchar(400) NOT NULL,
 107                    envelope_json nvarchar(max) NOT NULL,
 108                    created_at datetime2 NOT NULL DEFAULT SYSUTCDATETIME(),
 109                    expires_at datetime2 NOT NULL,
 110                    acked_at datetime2 NULL,
 111                    recovery_claimed bit NOT NULL DEFAULT 0
 112                );
 113                IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name = N'{IndexName(_options.MessageTable, "correlation_c
 114                    CREATE INDEX {Quote(IndexName(_options.MessageTable, "correlation_created"))}
 115                        ON {MessageTable} (correlation_id, created_at);
 116                IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name = N'{IndexName(_options.MessageTable, "expires")}' A
 117                    CREATE INDEX {Quote(IndexName(_options.MessageTable, "expires"))}
 118                        ON {MessageTable} (expires_at);
 119
 120                IF OBJECT_ID(N'{SubscriberTable}', N'U') IS NULL
 121                CREATE TABLE {SubscriberTable} (
 122                    correlation_id nvarchar(400) NOT NULL,
 123                    registration_id uniqueidentifier NOT NULL,
 124                    instance_id nvarchar(200) NOT NULL,
 125                    expires_at datetime2 NOT NULL,
 126                    PRIMARY KEY (correlation_id, registration_id)
 127                );
 128                IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name = N'{IndexName(_options.SubscriberTable, "expires")}
 129                    CREATE INDEX {Quote(IndexName(_options.SubscriberTable, "expires"))}
 130                        ON {SubscriberTable} (expires_at);
 131                """;
 132            await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 133            await transaction.CommitAsync(cancellationToken).ConfigureAwait(false);
 134            _created = true;
 135        }
 136        finally
 137        {
 138            _ensureGate.Release();
 139        }
 140    }
 141
 142    public async Task SaveRecoveryStateAsync(string correlationId, RecoveryState state, TimeSpan ttl, CancellationToken 
 143    {
 144        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 145        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 146        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.
 149        command.CommandText =
 150            $"""
 151            MERGE {RecoveryTable} WITH (HOLDLOCK) AS target
 152            USING (SELECT @correlation_id AS correlation_id, @registration_id AS registration_id) AS source
 153                ON target.correlation_id = source.correlation_id AND target.registration_id = source.registration_id
 154            WHEN MATCHED THEN
 155                UPDATE SET state_json = @state_json,
 156                           expires_at = {AddMilliseconds("@ttl_ms")},
 157                           registered_at = SYSUTCDATETIME()
 158            WHEN NOT MATCHED THEN
 159                INSERT (correlation_id, registration_id, state_json, expires_at, registered_at)
 160                VALUES (@correlation_id, @registration_id, @state_json, {AddMilliseconds("@ttl_ms")}, SYSUTCDATETIME());
 161            """;
 162        command.Parameters.AddWithValue("@correlation_id", correlationId);
 163        command.Parameters.AddWithValue("@registration_id", state.RegistrationId);
 164        command.Parameters.AddWithValue("@state_json", AsyncResponseJson.Serialize(state));
 165        command.Parameters.AddWithValue("@ttl_ms", (long)ttl.TotalMilliseconds);
 166        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 167    }
 168
 169    public async Task<IReadOnlyList<string>> LoadRecoveryStatesAsync(string correlationId, CancellationToken cancellatio
 170    {
 171        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 172        if (ShouldPrune(ref _lastRecoveryPruneTicks))
 173            await PruneExpiredRecoveryAsync(correlationId, cancellationToken).ConfigureAwait(false);
 174
 175        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 176        await using var command = connection.CreateCommand();
 177        command.CommandText =
 178            $"""
 179            SELECT state_json
 180            FROM {RecoveryTable}
 181            WHERE correlation_id = @correlation_id AND expires_at > SYSUTCDATETIME()
 182            ORDER BY registered_at;
 183            """;
 184        command.Parameters.AddWithValue("@correlation_id", correlationId);
 185
 186        var states = new List<string>();
 187        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 188        while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 189            states.Add(reader.GetString(0));
 190        return states;
 191    }
 192
 193    public async Task<bool> DeleteRecoveryStateAsync(string correlationId, Guid registrationId, CancellationToken cancel
 194    {
 195        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 196        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 197        await using var command = connection.CreateCommand();
 198        command.CommandText = $"DELETE FROM {RecoveryTable} WHERE correlation_id = @correlation_id AND registration_id =
 199        command.Parameters.AddWithValue("@correlation_id", correlationId);
 200        command.Parameters.AddWithValue("@registration_id", registrationId);
 201        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 202    }
 203
 204    public async IAsyncEnumerable<string> ScanRecoveryStateJsonAsync([System.Runtime.CompilerServices.EnumeratorCancella
 205    {
 206        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 207        await PruneExpiredRecoveryAsync(null, cancellationToken).ConfigureAwait(false);
 208
 209        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 210        await using var command = connection.CreateCommand();
 211        command.CommandText =
 212            $"""
 213            SELECT state_json
 214            FROM {RecoveryTable}
 215            WHERE expires_at > SYSUTCDATETIME()
 216            ORDER BY registered_at;
 217            """;
 218        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 219        while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 220            yield return reader.GetString(0);
 221    }
 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
 232        => AsyncResponseRetry.ExecuteAsync(
 233            token => InsertMessageOnceAsync(id, correlationId, envelopeJson, retention, token),
 234            IsTransient,
 235            _options.PublishMaxAttempts,
 236            _options.PublishRetryBaseDelay,
 237            _options.PublishRetryMaxDelay,
 238            cancellationToken);
 239
 240    private async Task<DateTimeOffset> InsertMessageOnceAsync(Guid id, string correlationId, string envelopeJson, TimeSp
 241    {
 242        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 243        if (ShouldPrune(ref _lastMessagePruneTicks))
 244            await PruneExpiredMessagesAsync(cancellationToken).ConfigureAwait(false);
 245
 246        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 247        await using var command = connection.CreateCommand();
 248        command.CommandText =
 249            $"""
 250            INSERT INTO {MessageTable} (id, correlation_id, envelope_json, expires_at)
 251            OUTPUT inserted.created_at
 252            SELECT @id, @correlation_id, @envelope_json, {AddMilliseconds("@retention_ms")}
 253            WHERE NOT EXISTS (SELECT 1 FROM {MessageTable} WITH (UPDLOCK, HOLDLOCK) WHERE id = @id);
 254            """;
 255        command.Parameters.AddWithValue("@id", id);
 256        command.Parameters.AddWithValue("@correlation_id", correlationId);
 257        command.Parameters.AddWithValue("@envelope_json", envelopeJson);
 258        command.Parameters.AddWithValue("@retention_ms", (long)retention.TotalMilliseconds);
 259
 260        object? createdAt = null;
 261        try
 262        {
 263            createdAt = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
 264        }
 265        catch (SqlException ex) when (ex.Number is PrimaryKeyViolation or UniqueIndexViolation)
 266        {
 267        }
 268
 269        if (createdAt is DateTime insertedCreatedAt)
 270            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.
 278        await using var lookup = connection.CreateCommand();
 279        lookup.CommandText = $"SELECT created_at FROM {MessageTable} WHERE id = @id;";
 280        lookup.Parameters.AddWithValue("@id", id);
 281        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.
 287        return existing is DateTime existingCreatedAt
 288            ? new DateTimeOffset(existingCreatedAt, TimeSpan.Zero)
 289            : throw new InvalidOperationException(
 290                $"SQL Server response insert for message {id} found no row after a duplicate: the original no longer exi
 291    }
 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    {
 301        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 302        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 303        await using var command = connection.CreateCommand();
 304        command.CommandText =
 305            $"""
 306            SELECT id, correlation_id, envelope_json, created_at, acked_at
 307            FROM {MessageTable}
 308            WHERE correlation_id = @correlation_id
 309              AND created_at >= @since
 310              AND expires_at > SYSUTCDATETIME()
 311              {(afterCreatedAtUtc is null ? "" : "AND (created_at > @after_created_at OR (created_at = @after_created_at
 312            ORDER BY created_at, id
 313            OFFSET 0 ROWS FETCH NEXT @limit ROWS ONLY;
 314            """;
 315        command.Parameters.AddWithValue("@correlation_id", correlationId);
 316        var sinceParameter = command.Parameters.Add("@since", SqlDbType.DateTime2);
 317        sinceParameter.Scale = 7;
 318        sinceParameter.Value = sinceUtc.UtcDateTime;
 319        command.Parameters.AddWithValue("@limit", batchSize);
 320        if (afterCreatedAtUtc is not null)
 321        {
 322            var cursorParameter = command.Parameters.Add("@after_created_at", SqlDbType.DateTime2);
 323            cursorParameter.Scale = 7;
 324            cursorParameter.Value = afterCreatedAtUtc.Value.UtcDateTime;
 325            command.Parameters.AddWithValue("@after_id", afterId ?? throw new ArgumentNullException(nameof(afterId)));
 326        }
 327
 328        var messages = new List<SqlServerChannelMessage>(batchSize);
 329        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 330        while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 331            messages.Add(new SqlServerChannelMessage(
 332                reader.GetGuid(0),
 333                reader.GetString(1),
 334                reader.GetString(2),
 335                new DateTimeOffset(reader.GetDateTime(3), TimeSpan.Zero),
 336                reader.IsDBNull(4) ? null : new DateTimeOffset(reader.GetDateTime(4), TimeSpan.Zero)));
 337        return messages;
 338    }
 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    {
 349        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 350        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 351        await using var command = connection.CreateCommand();
 352        command.CommandText =
 353            $"""
 354            UPDATE {MessageTable}
 355            SET acked_at = COALESCE(acked_at, SYSUTCDATETIME())
 356            OUTPUT inserted.id
 357            WHERE id = @id AND recovery_claimed = 0 AND expires_at > SYSUTCDATETIME();
 358            """;
 359        command.Parameters.AddWithValue("@id", messageId);
 360        var result = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
 361        return result is not null and not DBNull;
 362    }
 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    {
 372        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 373        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 374        await using var command = connection.CreateCommand();
 375        command.CommandText =
 376            $"""
 377            UPDATE {MessageTable}
 378            SET recovery_claimed = 1
 379            OUTPUT inserted.id
 380            WHERE id = @id AND acked_at IS NULL;
 381            """;
 382        command.Parameters.AddWithValue("@id", messageId);
 383        var result = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
 384        return result is not null and not DBNull;
 385    }
 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    {
 390        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 391        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 392        await using var command = connection.CreateCommand();
 393        command.CommandText = "SELECT SYSUTCDATETIME();";
 394        var result = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
 395        return result switch
 396        {
 397            DateTimeOffset dto => dto.ToUniversalTime(),
 398            DateTime dt => new DateTimeOffset(DateTime.SpecifyKind(dt, DateTimeKind.Utc), TimeSpan.Zero),
 399            _ => DateTimeOffset.UtcNow
 400        };
 401    }
 402
 403    public async Task<bool> IsMessageAcknowledgedAsync(Guid messageId, CancellationToken cancellationToken)
 404    {
 405        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 406        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 407        await using var command = connection.CreateCommand();
 408        command.CommandText =
 409            $"""
 410            SELECT CAST(CASE WHEN acked_at IS NOT NULL THEN 1 ELSE 0 END AS bit)
 411            FROM {MessageTable}
 412            WHERE id = @id AND expires_at > SYSUTCDATETIME();
 413            """;
 414        command.Parameters.AddWithValue("@id", messageId);
 415        var result = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
 416        return result is bool acknowledged && acknowledged;
 417    }
 418
 419    public async Task UpsertSubscriberAsync(string correlationId, Guid registrationId, string instanceId, TimeSpan ttl, 
 420    {
 421        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 422        if (ShouldPrune(ref _lastSubscriberPruneTicks))
 423            await PruneExpiredSubscribersAsync(correlationId, cancellationToken).ConfigureAwait(false);
 424
 425        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 426        await using var command = connection.CreateCommand();
 427        command.CommandText =
 428            $"""
 429            MERGE {SubscriberTable} WITH (HOLDLOCK) AS target
 430            USING (SELECT @correlation_id AS correlation_id, @registration_id AS registration_id) AS source
 431                ON target.correlation_id = source.correlation_id AND target.registration_id = source.registration_id
 432            WHEN MATCHED THEN
 433                UPDATE SET instance_id = @instance_id,
 434                           expires_at = {AddMilliseconds("@ttl_ms")}
 435            WHEN NOT MATCHED THEN
 436                INSERT (correlation_id, registration_id, instance_id, expires_at)
 437                VALUES (@correlation_id, @registration_id, @instance_id, {AddMilliseconds("@ttl_ms")});
 438            """;
 439        command.Parameters.AddWithValue("@correlation_id", correlationId);
 440        command.Parameters.AddWithValue("@registration_id", registrationId);
 441        command.Parameters.AddWithValue("@instance_id", instanceId);
 442        command.Parameters.AddWithValue("@ttl_ms", (long)ttl.TotalMilliseconds);
 443        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 444    }
 445
 446    public async Task HeartbeatSubscribersAsync(
 447        string instanceId,
 448        IReadOnlyList<(string CorrelationId, Guid RegistrationId)> registrations,
 449        TimeSpan ttl,
 450        CancellationToken cancellationToken)
 451    {
 452        if (registrations.Count == 0)
 453            return;
 454
 455        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 456        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;
 460        for (var offset = 0; offset < registrations.Count; offset += batchSize)
 461        {
 462            var count = Math.Min(batchSize, registrations.Count - offset);
 463            await using var command = connection.CreateCommand();
 464            var sourceRows = new string[count];
 465            for (var index = 0; index < count; index++)
 466            {
 467                var (correlationId, registrationId) = registrations[offset + index];
 468                sourceRows[index] = $"(@correlation_id_{index}, @registration_id_{index})";
 469                command.Parameters.AddWithValue($"@correlation_id_{index}", correlationId);
 470                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".
 477            command.CommandText =
 478                $"""
 479                MERGE {SubscriberTable} WITH (HOLDLOCK) AS target
 480                USING (VALUES {string.Join(", ", sourceRows)}) AS source (correlation_id, registration_id)
 481                    ON target.correlation_id = source.correlation_id AND target.registration_id = source.registration_id
 482                WHEN MATCHED THEN
 483                    UPDATE SET instance_id = @instance_id,
 484                               expires_at = {AddMilliseconds("@ttl_ms")}
 485                WHEN NOT MATCHED THEN
 486                    INSERT (correlation_id, registration_id, instance_id, expires_at)
 487                    VALUES (source.correlation_id, source.registration_id, @instance_id, {AddMilliseconds("@ttl_ms")});
 488                """;
 489            command.Parameters.AddWithValue("@instance_id", instanceId);
 490            command.Parameters.AddWithValue("@ttl_ms", (long)ttl.TotalMilliseconds);
 491            await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 492        }
 493    }
 494
 495    public async Task DeleteSubscriberAsync(string correlationId, Guid registrationId, CancellationToken cancellationTok
 496    {
 497        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 498        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 499        await using var command = connection.CreateCommand();
 500        command.CommandText = $"DELETE FROM {SubscriberTable} WHERE correlation_id = @correlation_id AND registration_id
 501        command.Parameters.AddWithValue("@correlation_id", correlationId);
 502        command.Parameters.AddWithValue("@registration_id", registrationId);
 503        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 504    }
 505
 506    public async Task<long> CountActiveSubscribersAsync(string correlationId, CancellationToken cancellationToken)
 507    {
 508        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 509        if (ShouldPrune(ref _lastSubscriberPruneTicks))
 510            await PruneExpiredSubscribersAsync(correlationId, cancellationToken).ConfigureAwait(false);
 511
 512        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 513        await using var command = connection.CreateCommand();
 514        command.CommandText =
 515            $"""
 516            SELECT COUNT_BIG(*)
 517            FROM {SubscriberTable}
 518            WHERE correlation_id = @correlation_id AND expires_at > SYSUTCDATETIME();
 519            """;
 520        command.Parameters.AddWithValue("@correlation_id", correlationId);
 521        var result = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
 522        return result is long count ? count : 0L;
 523    }
 524
 525    private async Task PruneExpiredRecoveryAsync(string? correlationId, CancellationToken cancellationToken)
 526    {
 527        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 528        await using var command = connection.CreateCommand();
 529        command.CommandText = correlationId is null
 530            ? $"DELETE FROM {RecoveryTable} WHERE expires_at <= SYSUTCDATETIME();"
 531            : $"DELETE FROM {RecoveryTable} WHERE correlation_id = @correlation_id AND expires_at <= SYSUTCDATETIME();";
 532        if (correlationId is not null)
 533            command.Parameters.AddWithValue("@correlation_id", correlationId);
 534        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 535    }
 536
 537    private async Task PruneExpiredMessagesAsync(CancellationToken cancellationToken)
 538    {
 539        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 540        await using var command = connection.CreateCommand();
 541        command.CommandText = $"DELETE FROM {MessageTable} WHERE expires_at <= SYSUTCDATETIME();";
 542        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 543    }
 544
 545    private async Task PruneExpiredSubscribersAsync(string? correlationId, CancellationToken cancellationToken)
 546    {
 547        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 548        await using var command = connection.CreateCommand();
 549        command.CommandText = correlationId is null
 550            ? $"DELETE FROM {SubscriberTable} WHERE expires_at <= SYSUTCDATETIME();"
 551            : $"DELETE FROM {SubscriberTable} WHERE correlation_id = @correlation_id AND expires_at <= SYSUTCDATETIME();
 552        if (correlationId is not null)
 553            command.Parameters.AddWithValue("@correlation_id", correlationId);
 554        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 555    }
 556
 557    private async Task<SqlConnection> OpenConnectionAsync(CancellationToken cancellationToken)
 558    {
 559        var connection = new SqlConnection(_connectionString);
 560        try
 561        {
 562            await connection.OpenAsync(cancellationToken).ConfigureAwait(false);
 563            return connection;
 564        }
 565        catch
 566        {
 567            await connection.DisposeAsync().ConfigureAwait(false);
 568            throw;
 569        }
 570    }
 571
 572    public static void ValidateIdentifier(string? value, string name)
 573    {
 574        if (string.IsNullOrWhiteSpace(value))
 575            throw new InvalidOperationException($"{nameof(SqlServerAsyncResponseChannelOptions)}.{name} must be configur
 576        if (!IsIdentifier(value))
 577            throw new InvalidOperationException(
 578                $"{nameof(SqlServerAsyncResponseChannelOptions)}.{name} '{value}' must be a simple SQL Server identifier
 579    }
 580
 581    private static bool IsIdentifier(string value)
 582    {
 583        if (value.Length == 0 || !(char.IsAsciiLetter(value[0]) || value[0] == '_'))
 584            return false;
 585
 586        foreach (var c in value)
 587        {
 588            if (!(char.IsAsciiLetterOrDigit(c) || c == '_'))
 589                return false;
 590        }
 591
 592        return true;
 593    }
 594
 595    private static string Quote(string identifier) => "[" + identifier + "]";
 596
 597    private static string IndexName(string table, string suffix)
 598    {
 599        var name = $"{table}_{suffix}_idx";
 600        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)
 610        => $"DATEADD(SECOND, CAST({parameterName} / 1000 AS int), DATEADD(MILLISECOND, CAST({parameterName} % 1000 AS in
 611
 612    internal static bool IsTransient(Exception exception)
 613        => exception is not OperationCanceledException
 614           && (exception is SqlException sqlException && SqlServerTransientFaults.IsTransient(sqlException)
 615               || 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)
 623        => $"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    {
 632        var interval = _options.PruneInterval;
 633        if (interval <= TimeSpan.Zero)
 634            return true;
 635
 636        var now = DateTime.UtcNow.Ticks;
 637        var last = Interlocked.Read(ref lastTicks);
 638        return now - last >= interval.Ticks
 639            && 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{
 3650    private static readonly HashSet<int> TransientErrorNumbers =
 3651    [
 3652        -2,    // client-side command timeout
 3653        20,    // instance does not support encryption
 3654        64,    // connection lost during login
 3655        121,   // transport semaphore timeout
 3656        233,   // no process on the other end of the pipe
 3657        997,   // overlapped I/O in progress
 3658        1204,  // lock resources exhausted
 3659        1205,  // deadlock victim
 3660        1222,  // lock request timeout
 3661        4060,  // database unavailable
 3662        4221,  // readable secondary timeout
 3663        10053, // transport-level connection abort
 3664        10054, // transport-level connection reset
 3665        10060, // network unreachable / connect timeout
 3666        10928, // Azure SQL resource limit reached
 3667        10929, // Azure SQL minimum guarantee exceeded
 3668        40143, // Azure SQL connection failure
 3669        40197, // Azure SQL service processing error
 3670        40501, // Azure SQL service busy
 3671        40540, // Azure SQL service unavailable
 3672        40613, // Azure SQL database unavailable
 3673        49918, // cannot process request, not enough resources
 3674        49919, // cannot process create/update request
 3675        49920  // cannot process request, too many operations
 3676    ];
 677
 678    public static bool IsTransient(SqlException exception)
 679    {
 2680        if (exception.Class >= 20)
 2681            return true;
 682
 2683        foreach (SqlError error in exception.Errors)
 684        {
 2685            if (TransientErrorNumbers.Contains(error.Number))
 2686                return true;
 687        }
 688
 2689        return TransientErrorNumbers.Contains(exception.Number);
 2690    }
 691}
Generated by: ReportGenerator 5.5.10.0
08/03/2026 - 08:59:44
GitHub | reportgenerator.io

Methods/Properties

.cctor()
IsTransient(Microsoft.Data.SqlClient.SqlException)