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

Information
Class: AsyncResponse.Channels.SqlServer.SqlServerChannelSql
Assembly: AsyncResponse.Channels.SqlServer
File(s): /_/src/Channels/AsyncResponse.Channels.SqlServer/SqlServerChannelSql.cs
Line coverage
98%
Covered lines: 562
Uncovered lines: 8
Coverable lines: 570
Total lines: 959
Line coverage: 98.5%
Branch coverage
89%
Covered branches: 109
Total branches: 122
Branch coverage: 89.3%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%11100%
get_Schema()100%11100%
get_RecoveryTable()100%11100%
get_MessageTable()100%11100%
get_SubscriberTable()100%11100%
get_AckSequence()100%11100%
get_AckSequenceName()100%11100%
EnsureCreatedAsync()83.33%6698.95%
VerifyRelationsAsync(...)100%11100%
ExpectedObjects()100%11100%
ValidateManagedSchemaAsync()68.75%161696.66%
SaveRecoveryStateAsync()100%11100%
LoadRecoveryStatesAsync()100%66100%
DeleteRecoveryStateAsync()100%11100%
ScanRecoveryStateJsonAsync()100%1010100%
InsertMessageAsync(...)100%11100%
InsertMessageOnceAsync()91.66%121289.47%
LoadMessagesAsync()83.33%66100%
LoadMessagesByIdAsync()75%4495.23%
ReadMessagesAsync()100%88100%
TryClaimForDeliveryAsync()100%22100%
TryClaimForRecoveryAsync()100%22100%
GetSubscriptionStartAsync()100%22100%
GetServerTimeUtcAsync()50%4483.33%
IsMessageAcknowledgedAsync()50%22100%
UpsertSubscriberAsync()100%22100%
HeartbeatSubscribersAsync()83.33%6696.66%
DeleteSubscriberAsync()100%11100%
CountActiveSubscribersAsync()75%44100%
ExpiredPruneSql(...)100%11100%
PruneExpiredRecoveryAsync()100%44100%
PruneExpiredMessagesAsync()100%11100%
PruneExpiredSubscribersAsync()75%44100%
OpenConnectionAsync()100%11100%
ValidateIdentifier(...)100%66100%
IsIdentifier(...)100%1212100%
Quote(...)100%11100%
SequenceName(...)100%11100%
IndexName(...)100%11100%
ValidateNamePlan(...)100%11100%
AddMilliseconds(...)100%11100%
IsTransient(...)100%11100%
SchemaLockResource(...)100%11100%
ShouldPrune(...)100%44100%

File(s)

/_/src/Channels/AsyncResponse.Channels.SqlServer/SqlServerChannelSql.cs

#LineLine coverage
 1using AsyncResponse.Internal;
 2using Microsoft.Data.SqlClient;
 3using System.Data;
 4
 5namespace AsyncResponse.Channels.SqlServer;
 6
 7/// <summary>One stored response envelope row/document as the channel store returns it.</summary>
 8/// <remarks>
 9/// <c>EnvelopeJson</c> is the stored envelope, or <c>null</c> for a row the dispatch sweep loaded header-only (an
 10/// already-acknowledged row — see <see cref="SqlServerChannelSql.LoadMessagesAsync"/>); the
 11/// sweep hydrates the few such rows it still has to deliver through
 12/// <see cref="SqlServerChannelSql.LoadMessagesByIdAsync"/> before handing them to a waiter.
 13/// </remarks>
 14internal readonly record struct SqlServerChannelMessage(
 15    Guid Id,
 16    string CorrelationId,
 17    string? EnvelopeJson,
 18    DateTimeOffset CreatedAtUtc,
 19    DateTimeOffset? AckedAtUtc = null,
 20    long? AckedSeq = null);
 21
 22/// <summary>SQL helper for the SQL Server channel tables.</summary>
 23internal sealed class SqlServerChannelSql
 24{
 25    // SQL Server duplicate-key error numbers: 2627 = PRIMARY KEY/UNIQUE constraint violation,
 26    // 2601 = unique index violation. Retried idempotent inserts treat them as success.
 27    private const int PrimaryKeyViolation = 2627;
 28    private const int UniqueIndexViolation = 2601;
 29
 30    private readonly string _connectionString;
 31    private readonly SqlServerAsyncResponseChannelOptions _options;
 44932    private readonly SemaphoreSlim _ensureGate = new(1, 1);
 33    private bool _created;
 34    private long _lastRecoveryPruneTicks;
 35    private long _lastMessagePruneTicks;
 36    private long _lastSubscriberPruneTicks;
 37
 44938    public SqlServerChannelSql(Microsoft.Extensions.Options.IOptions<SqlServerAsyncResponseChannelOptions> options)
 39    {
 44940        _options = options.Value;
 44941        _options.Validate();
 44942        _connectionString = _options.ConnectionString!;
 43
 44944        Schema = Quote(_options.SchemaName);
 44945        RecoveryTable = $"{Schema}.{Quote(_options.RecoveryStateTable)}";
 44946        MessageTable = $"{Schema}.{Quote(_options.MessageTable)}";
 44947        SubscriberTable = $"{Schema}.{Quote(_options.SubscriberTable)}";
 44948        AckSequenceName = SequenceName(_options.MessageTable);
 44949        AckSequence = $"{Schema}.{Quote(AckSequenceName)}";
 44950    }
 51
 218152    public string Schema { get; }
 254953    public string RecoveryTable { get; }
 1209054    public string MessageTable { get; }
 606955    public string SubscriberTable { get; }
 56
 57    /// <summary>
 58    /// Qualified name of the monotonic ack sequence. Delivery claims and subscription
 59    /// registrations draw from this ONE sequence, giving <c>acked_seq</c> and a subscription's
 60    /// start position a total order no pair of same-tick timestamps has.
 61    /// </summary>
 142262    public string AckSequence { get; }
 63
 123564    private string AckSequenceName { get; }
 65
 66    public async Task EnsureCreatedAsync(CancellationToken cancellationToken = default)
 67    {
 1316168        if (_created)
 1272569            return;
 70
 43671        if (!_options.AutoCreateSchema)
 72        {
 73            // Manually managed schemas get a one-time validation instead of DDL: 1.0.0 added
 74            // acked_seq and its sequence, which waiter registration and delivery claims require
 75            // unconditionally — without this check an un-migrated schema fails later with a raw
 76            // "invalid column name" mid-operation instead of an actionable startup error carrying
 77            // the exact migration.
 5278            await ValidateManagedSchemaAsync(cancellationToken).ConfigureAwait(false);
 279            return;
 80        }
 81
 38482        await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 83        try
 84        {
 38485            if (_created)
 086                return;
 87
 38488            await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 38289            await using var transaction = (SqlTransaction)await connection.BeginTransactionAsync(cancellationToken).Conf
 90
 91            // Serialize schema creation across processes. The IF-NOT-EXISTS guards are not atomic
 92            // against a concurrent create of the same object: two instances starting together both
 93            // pass the existence check and collide on the catalog (error 2714/2627). A
 94            // transaction-scoped application lock (keyed by schema, shared with the transport store)
 95            // lets one instance build the schema while the rest wait and then find it already present.
 38296            await using (var lockCommand = connection.CreateCommand())
 97            {
 38298                lockCommand.Transaction = transaction;
 38299                lockCommand.CommandText =
 382100                    """
 382101                    DECLARE @lock_result int;
 382102                    EXEC @lock_result = sp_getapplock
 382103                        @Resource = @lock_resource,
 382104                        @LockMode = 'Exclusive',
 382105                        @LockOwner = 'Transaction',
 382106                        @LockTimeout = 60000;
 382107                    IF @lock_result < 0
 382108                        THROW 51000, N'Failed to acquire the AsyncResponse DDL application lock.', 1;
 382109                    """;
 382110                lockCommand.Parameters.AddWithValue("@lock_resource", SchemaLockResource(_options.SchemaName));
 382111                await lockCommand.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 112            }
 113
 382114            await using var command = connection.CreateCommand();
 382115            command.Transaction = transaction;
 382116            command.CommandText =
 382117                $"""
 382118                IF SCHEMA_ID(N'{_options.SchemaName}') IS NULL
 382119                    EXEC(N'CREATE SCHEMA {Schema}');
 382120
 382121                IF OBJECT_ID(N'{RecoveryTable}', N'U') IS NULL
 382122                CREATE TABLE {RecoveryTable} (
 382123                    correlation_id nvarchar(400) COLLATE Latin1_General_100_BIN2 NOT NULL,
 382124                    registration_id uniqueidentifier NOT NULL,
 382125                    state_json nvarchar(max) NOT NULL,
 382126                    expires_at datetime2 NOT NULL,
 382127                    registered_at datetime2 NOT NULL DEFAULT SYSUTCDATETIME(),
 382128                    PRIMARY KEY (correlation_id, registration_id)
 382129                );
 382130                IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name = N'{IndexName(_options.RecoveryStateTable, "expires
 382131                    CREATE INDEX {Quote(IndexName(_options.RecoveryStateTable, "expires"))}
 382132                        ON {RecoveryTable} (expires_at);
 382133
 382134                IF OBJECT_ID(N'{MessageTable}', N'U') IS NULL
 382135                CREATE TABLE {MessageTable} (
 382136                    id uniqueidentifier NOT NULL PRIMARY KEY NONCLUSTERED,
 382137                    correlation_id nvarchar(400) COLLATE Latin1_General_100_BIN2 NOT NULL,
 382138                    envelope_json nvarchar(max) NOT NULL,
 382139                    created_at datetime2 NOT NULL DEFAULT SYSUTCDATETIME(),
 382140                    expires_at datetime2 NOT NULL,
 382141                    acked_at datetime2 NULL,
 382142                    acked_seq bigint NULL,
 382143                    recovery_claimed bit NOT NULL DEFAULT 0
 382144                );
 382145                IF COL_LENGTH(N'{MessageTable}', N'acked_seq') IS NULL
 382146                    ALTER TABLE {MessageTable} ADD acked_seq bigint NULL;
 382147                IF NOT EXISTS (SELECT 1 FROM sys.sequences WHERE name = N'{AckSequenceName}' AND schema_id = SCHEMA_ID(N
 382148                    CREATE SEQUENCE {AckSequence} AS bigint START WITH 1;
 382149                IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name = N'{IndexName(_options.MessageTable, "correlation_c
 382150                    CREATE INDEX {Quote(IndexName(_options.MessageTable, "correlation_created"))}
 382151                        ON {MessageTable} (correlation_id, created_at);
 382152                IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name = N'{IndexName(_options.MessageTable, "expires")}' A
 382153                    CREATE INDEX {Quote(IndexName(_options.MessageTable, "expires"))}
 382154                        ON {MessageTable} (expires_at);
 382155
 382156                IF OBJECT_ID(N'{SubscriberTable}', N'U') IS NULL
 382157                CREATE TABLE {SubscriberTable} (
 382158                    correlation_id nvarchar(400) COLLATE Latin1_General_100_BIN2 NOT NULL,
 382159                    registration_id uniqueidentifier NOT NULL,
 382160                    instance_id nvarchar(200) NOT NULL,
 382161                    expires_at datetime2 NOT NULL,
 382162                    PRIMARY KEY (correlation_id, registration_id)
 382163                );
 382164                IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name = N'{IndexName(_options.SubscriberTable, "expires")}
 382165                    CREATE INDEX {Quote(IndexName(_options.SubscriberTable, "expires"))}
 382166                        ON {SubscriberTable} (expires_at);
 382167                """;
 168            try
 169            {
 382170                await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 378171            }
 4172            catch (SqlException ex)
 173            {
 174                // The batch can break BEFORE the verification below ever runs: a name held by
 175                // another component's table suppresses the guarded CREATE and the statements that
 176                // follow (an index over columns that table lacks, the acked_seq ALTER) hit the
 177                // wrong table, and a name held by a view fails outright with error 2714. Run the
 178                // very same catalog checks now — on a fresh connection, since the objects in
 179                // question are somebody else's and already committed — so the operator gets the
 180                // precise reason instead of a raw provider error.
 4181                await SqlServerRelationVerifier.ThrowDiagnosedCollisionAsync(
 4182                    OpenConnectionAsync,
 4183                    ex,
 4184                    _options.SchemaName,
 4185                    "channel",
 4186                    ExpectedObjects(),
 4187                    cancellationToken).ConfigureAwait(false);
 1188                throw;
 189            }
 190
 378191            await transaction.CommitAsync(cancellationToken).ConfigureAwait(false);
 192
 193            // Verified AFTER the commit, on the same connection but outside the transaction. The
 194            // checks read the catalog, and a transaction that has just run DDL still holds
 195            // schema-modification locks — catalog reads under those deadlock (error 1205) against
 196            // this store's own live traffic, which is already polling by the time a later
 197            // EnsureCreated re-runs. Correctness does not need the transaction: the application
 198            // lock serialized the DDL, and what these checks look for is somebody ELSE'S committed
 199            // object occupying a name, never our own uncommitted work.
 378200            await VerifyRelationsAsync(connection, transaction: null, cancellationToken).ConfigureAwait(false);
 377201            _created = true;
 377202        }
 203        finally
 204        {
 384205            _ensureGate.Release();
 206        }
 13104207    }
 208
 209
 210    /// <summary>
 211    /// Post-DDL catalog verification, inside the DDL transaction (and therefore under the shared
 212    /// application lock). The existence guards above only ask "is there a user table with this
 213    /// name": a name held by another AsyncResponse component's table makes them skip creation
 214    /// silently, and a name held by a view or synonym makes the CREATE fail with raw error 2714.
 215    /// </summary>
 216    private Task VerifyRelationsAsync(SqlConnection connection, SqlTransaction? transaction, CancellationToken cancellat
 382217        => SqlServerRelationVerifier.VerifyAsync(
 382218            connection,
 382219            transaction,
 382220            _options.SchemaName,
 382221            "channel",
 382222            ExpectedObjects(),
 382223            cancellationToken);
 224
 225    /// <summary>The catalog shape this store's DDL intends — the single source for both the
 226    /// post-DDL verification and the failed-batch diagnosis.</summary>
 227    /// <remarks>A bare <c>datetime2</c> declaration is <c>datetime2(7)</c>; the expected types
 228    /// state the scale, because a reduced-scale column rounds the timestamps on store rather than
 229    /// merely displaying them coarsely.</remarks>
 230    private SqlServerRelationVerifier.ExpectedObject[] ExpectedObjects() =>
 388231            [
 388232                new(_options.RecoveryStateTable, SqlServerObjectKind.Table,
 388233                [
 388234                    new("correlation_id", "nvarchar(400)", Nullable: false, RequiresBinaryCollation: true),
 388235                    new("registration_id", "uniqueidentifier", Nullable: false),
 388236                    new("state_json", "nvarchar(max)", Nullable: false),
 388237                    new("expires_at", "datetime2(7)", Nullable: false),
 388238                    new("registered_at", "datetime2(7)", Nullable: false, DefaultExpression: "(sysutcdatetime())")
 388239                ],
 388240                PrimaryKey: ["correlation_id", "registration_id"]),
 388241                new(_options.MessageTable, SqlServerObjectKind.Table,
 388242                [
 388243                    new("id", "uniqueidentifier", Nullable: false),
 388244                    new("correlation_id", "nvarchar(400)", Nullable: false, RequiresBinaryCollation: true),
 388245                    new("envelope_json", "nvarchar(max)", Nullable: false),
 388246                    new("created_at", "datetime2(7)", Nullable: false, DefaultExpression: "(sysutcdatetime())"),
 388247                    new("expires_at", "datetime2(7)", Nullable: false),
 388248                    new("acked_at", "datetime2(7)", Nullable: true),
 388249                    new("acked_seq", "bigint", Nullable: true),
 388250                    new("recovery_claimed", "bit", Nullable: false, DefaultExpression: "((0))")
 388251                ],
 388252                PrimaryKey: ["id"]),
 388253                new(_options.SubscriberTable, SqlServerObjectKind.Table,
 388254                [
 388255                    new("correlation_id", "nvarchar(400)", Nullable: false, RequiresBinaryCollation: true),
 388256                    new("registration_id", "uniqueidentifier", Nullable: false),
 388257                    new("instance_id", "nvarchar(200)", Nullable: false),
 388258                    new("expires_at", "datetime2(7)", Nullable: false)
 388259                ],
 388260                PrimaryKey: ["correlation_id", "registration_id"]),
 388261                new(AckSequenceName, SqlServerObjectKind.Sequence),
 388262                new(IndexName(_options.RecoveryStateTable, "expires"), SqlServerObjectKind.Index,
 388263                    OwningTable: _options.RecoveryStateTable, KeyColumns: ["expires_at"]),
 388264                new(IndexName(_options.MessageTable, "correlation_created"), SqlServerObjectKind.Index,
 388265                    OwningTable: _options.MessageTable, KeyColumns: ["correlation_id", "created_at"]),
 388266                new(IndexName(_options.MessageTable, "expires"), SqlServerObjectKind.Index,
 388267                    OwningTable: _options.MessageTable, KeyColumns: ["expires_at"]),
 388268                new(IndexName(_options.SubscriberTable, "expires"), SqlServerObjectKind.Index,
 388269                    OwningTable: _options.SubscriberTable, KeyColumns: ["expires_at"])
 388270            ];
 271
 272    private async Task ValidateManagedSchemaAsync(CancellationToken cancellationToken)
 273    {
 52274        await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 275        try
 276        {
 52277            if (_created)
 0278                return;
 279
 52280            await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 281            bool hasColumn;
 282            bool hasSequence;
 283            // The probe's command and reader are scoped so they are disposed before the relation
 284            // verification below reuses this connection — no MARS, one active command at a time.
 10285            await using (var command = connection.CreateCommand())
 286            {
 10287                command.CommandText =
 10288                    $"""
 10289                    SELECT
 10290                      CASE WHEN COL_LENGTH(N'{MessageTable}', N'acked_seq') IS NOT NULL THEN 1 ELSE 0 END,
 10291                      CASE WHEN EXISTS (SELECT 1 FROM sys.sequences WHERE name = N'{AckSequenceName}' AND schema_id = SC
 10292                    """;
 10293                await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 10294                await reader.ReadAsync(cancellationToken).ConfigureAwait(false);
 10295                hasColumn = reader.GetInt32(0) == 1;
 10296                hasSequence = reader.GetInt32(1) == 1;
 10297            }
 10298            if (!hasColumn || !hasSequence)
 299            {
 6300                throw new InvalidOperationException(
 6301                    $"The SQL Server channel schema is managed manually (AutoCreateSchema = false) but is missing " +
 6302                    $"objects this version requires: " +
 6303                    $"{(hasColumn ? "" : $"column {MessageTable}.acked_seq")}{(!hasColumn && !hasSequence ? " and " : ""
 6304                    $"Apply the migration and restart: " +
 6305                    $"IF COL_LENGTH(N'{MessageTable}', N'acked_seq') IS NULL ALTER TABLE {MessageTable} ADD acked_seq bi
 6306                    $"IF NOT EXISTS (SELECT 1 FROM sys.sequences WHERE name = N'{AckSequenceName}' AND schema_id = SCHEM
 6307                    "See docs/sqlserver.md, section 'Upgrading a manually managed schema'.");
 308            }
 309
 310            // Full relation verification on the managed path too (transport/flow-store parity):
 311            // an operator-provisioned table with the wrong shape — a case-insensitive
 312            // correlation_id collation above all, under which `=` pads trailing spaces and
 313            // cross-routes responses — previously passed startup here and failed silently at
 314            // runtime, which is exactly what verification exists to catch.
 4315            await VerifyRelationsAsync(connection, transaction: null, cancellationToken).ConfigureAwait(false);
 316
 2317            _created = true;
 2318        }
 319        finally
 320        {
 52321            _ensureGate.Release();
 322        }
 2323    }
 324
 325    public async Task SaveRecoveryStateAsync(string correlationId, RecoveryState state, TimeSpan ttl, CancellationToken 
 326    {
 474327        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 474328        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 472329        await using var command = connection.CreateCommand();
 330        // MERGE WITH (HOLDLOCK) makes the match check and insert atomic — the SQL Server equivalent
 331        // of PostgreSQL's INSERT ... ON CONFLICT DO UPDATE for the (correlation_id, registration_id) key.
 472332        command.CommandText =
 472333            $"""
 472334            MERGE {RecoveryTable} WITH (HOLDLOCK) AS target
 472335            USING (SELECT @correlation_id AS correlation_id, @registration_id AS registration_id) AS source
 472336                ON target.correlation_id = source.correlation_id AND target.registration_id = source.registration_id
 472337            WHEN MATCHED THEN
 472338                UPDATE SET state_json = @state_json,
 472339                           expires_at = {AddMilliseconds("@ttl_ms")},
 472340                           registered_at = SYSUTCDATETIME()
 472341            WHEN NOT MATCHED THEN
 472342                INSERT (correlation_id, registration_id, state_json, expires_at, registered_at)
 472343                VALUES (@correlation_id, @registration_id, @state_json, {AddMilliseconds("@ttl_ms")}, SYSUTCDATETIME());
 472344            """;
 472345        command.Parameters.AddWithValue("@correlation_id", correlationId);
 472346        command.Parameters.AddWithValue("@registration_id", state.RegistrationId);
 472347        command.Parameters.AddWithValue("@state_json", AsyncResponseJson.Serialize(state));
 472348        command.Parameters.AddWithValue("@ttl_ms", (long)ttl.TotalMilliseconds);
 472349        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 472350    }
 351
 352    public async Task<IReadOnlyList<string>> LoadRecoveryStatesAsync(string correlationId, CancellationToken cancellatio
 353    {
 37354        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 37355        if (ShouldPrune(ref _lastRecoveryPruneTicks))
 37356            await PruneExpiredRecoveryAsync(correlationId, cancellationToken).ConfigureAwait(false);
 357
 37358        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 37359        await using var command = connection.CreateCommand();
 37360        command.CommandText =
 37361            $"""
 37362            SELECT state_json
 37363            FROM {RecoveryTable}
 37364            WHERE correlation_id = @correlation_id AND expires_at > SYSUTCDATETIME()
 37365            ORDER BY registered_at;
 37366            """;
 37367        command.Parameters.AddWithValue("@correlation_id", correlationId);
 368
 37369        var states = new List<string>();
 37370        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 62371        while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 25372            states.Add(reader.GetString(0));
 37373        return states;
 37374    }
 375
 376    public async Task<bool> DeleteRecoveryStateAsync(string correlationId, Guid registrationId, CancellationToken cancel
 377    {
 473378        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 473379        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 473380        await using var command = connection.CreateCommand();
 473381        command.CommandText = $"DELETE FROM {RecoveryTable} WHERE correlation_id = @correlation_id AND registration_id =
 473382        command.Parameters.AddWithValue("@correlation_id", correlationId);
 473383        command.Parameters.AddWithValue("@registration_id", registrationId);
 473384        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 473385    }
 386
 387    public async IAsyncEnumerable<string> ScanRecoveryStateJsonAsync([System.Runtime.CompilerServices.EnumeratorCancella
 388    {
 1389        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 1390        await PruneExpiredRecoveryAsync(null, cancellationToken).ConfigureAwait(false);
 391
 1392        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1393        await using var command = connection.CreateCommand();
 1394        command.CommandText =
 1395            $"""
 1396            SELECT state_json
 1397            FROM {RecoveryTable}
 1398            WHERE expires_at > SYSUTCDATETIME()
 1399            ORDER BY registered_at;
 1400            """;
 1401        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 2402        while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 1403            yield return reader.GetString(0);
 1404    }
 405
 406    /// <summary>
 407    /// Inserts a response envelope row. The caller supplies the message id so the insert is
 408    /// idempotent under retry — a duplicate insert (lost WHERE NOT EXISTS race or an outer retry)
 409    /// is treated as success, so a retried publish never duplicates a response. Returns the
 410    /// same-process fast-path message carrying the row's server-stamped <c>created_at</c> — and,
 411    /// on a duplicate, the ORIGINAL row's settlement columns, so the fast path compares against
 412    /// subscription watermarks exactly as the sweep does (a fabricated null <c>acked_at</c>
 413    /// replayed an already-consumed response to a waiter registered after the ack).
 414    /// </summary>
 415    public Task<SqlServerChannelMessage> InsertMessageAsync(Guid id, string correlationId, string envelopeJson, TimeSpan
 644416        => AsyncResponseRetry.ExecuteAsync(
 644417            token => InsertMessageOnceAsync(id, correlationId, envelopeJson, retention, token),
 644418            IsTransient,
 644419            _options.PublishMaxAttempts,
 644420            _options.PublishRetryBaseDelay,
 644421            _options.PublishRetryMaxDelay,
 644422            cancellationToken);
 423
 424    private async Task<SqlServerChannelMessage> InsertMessageOnceAsync(Guid id, string correlationId, string envelopeJso
 425    {
 644426        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 644427        if (ShouldPrune(ref _lastMessagePruneTicks))
 644428            await PruneExpiredMessagesAsync(cancellationToken).ConfigureAwait(false);
 429
 644430        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 644431        await using var command = connection.CreateCommand();
 644432        command.CommandText =
 644433            $"""
 644434            INSERT INTO {MessageTable} (id, correlation_id, envelope_json, expires_at)
 644435            OUTPUT inserted.created_at
 644436            SELECT @id, @correlation_id, @envelope_json, {AddMilliseconds("@retention_ms")}
 644437            WHERE NOT EXISTS (SELECT 1 FROM {MessageTable} WITH (UPDLOCK, HOLDLOCK) WHERE id = @id);
 644438            """;
 644439        command.Parameters.AddWithValue("@id", id);
 644440        command.Parameters.AddWithValue("@correlation_id", correlationId);
 644441        command.Parameters.AddWithValue("@envelope_json", envelopeJson);
 644442        command.Parameters.AddWithValue("@retention_ms", (long)retention.TotalMilliseconds);
 443
 644444        object? createdAt = null;
 445        try
 446        {
 644447            createdAt = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
 644448        }
 0449        catch (SqlException ex) when (ex.Number is PrimaryKeyViolation or UniqueIndexViolation)
 450        {
 0451        }
 452
 644453        if (createdAt is DateTime insertedCreatedAt)
 640454            return new SqlServerChannelMessage(id, correlationId, envelopeJson, new DateTimeOffset(insertedCreatedAt, Ti
 455
 456        // Duplicate insert (WHERE NOT EXISTS suppressed it, or the key-violation race lost):
 457        // return the original row with its server-stamped created_at AND its settlement columns,
 458        // so the same-process fast path compares against the watermark exactly as the sweep does
 459        // (a fabricated null acked_at replayed an already-consumed response to a waiter registered
 460        // after the ack). This fallback is a SEPARATE statement, so a concurrent same-id publish
 461        // is resolved here deterministically: the HOLDLOCK range lock on the first statement
 462        // serializes against the competing insert, and this second statement reads its own fresh
 463        // snapshot/locks and sees the committed row.
 4464        await using var lookup = connection.CreateCommand();
 4465        lookup.CommandText = $"SELECT created_at, acked_at, acked_seq FROM {MessageTable} WHERE id = @id;";
 4466        lookup.Parameters.AddWithValue("@id", id);
 4467        await using var existing = await lookup.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 4468        if (await existing.ReadAsync(cancellationToken).ConfigureAwait(false))
 469        {
 4470            return new SqlServerChannelMessage(
 4471                id,
 4472                correlationId,
 4473                envelopeJson,
 4474                new DateTimeOffset(existing.GetDateTime(0), TimeSpan.Zero),
 4475                existing.IsDBNull(1) ? null : new DateTimeOffset(existing.GetDateTime(1), TimeSpan.Zero),
 4476                existing.IsDBNull(2) ? null : existing.GetInt64(2));
 477        }
 478
 479        // A missing row means the idempotent duplicate's original is already gone (pruned
 480        // mid-publish): the message is not persisted, so reporting success with a fabricated
 481        // app-clock timestamp would both lie about persistence and feed a client clock into the
 482        // server-clock watermark. Fail instead, so the publisher's error handling runs.
 0483        throw new InvalidOperationException(
 0484            $"SQL Server response insert for message {id} found no row after a duplicate: the original no longer exists 
 644485    }
 486
 487    public async Task<IReadOnlyList<SqlServerChannelMessage>> LoadMessagesAsync(
 488        string correlationId,
 489        DateTimeOffset sinceUtc,
 490        int batchSize,
 491        DateTimeOffset? afterCreatedAtUtc,
 492        Guid? afterId,
 493        CancellationToken cancellationToken)
 494    {
 6353495        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 6343496        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 6339497        await using var command = connection.CreateCommand();
 498        // The envelope travels only for rows nobody has acknowledged yet. Acknowledged rows are
 499        // the consumed history the sweep re-reads on every tick (they stay in the result set so a
 500        // fan-out waiter in ANOTHER process still receives them): shipping their bodies with each
 501        // sweep made a long-lived progress subscription's cost grow with its whole retained
 502        // history. The shared sweep fetches the envelope by id for the rare acknowledged row a
 503        // live subscription has not seen.
 6339504        command.CommandText =
 6339505            $"""
 6339506            SELECT id, correlation_id, CASE WHEN acked_at IS NULL THEN envelope_json END, created_at, acked_at, acked_se
 6339507            FROM {MessageTable}
 6339508            WHERE correlation_id = @correlation_id
 6339509              AND created_at >= @since
 6339510              AND expires_at > SYSUTCDATETIME()
 6339511              {(afterCreatedAtUtc is null ? "" : "AND (created_at > @after_created_at OR (created_at = @after_created_at
 6339512            ORDER BY created_at, id
 6339513            OFFSET 0 ROWS FETCH NEXT @limit ROWS ONLY;
 6339514            """;
 6339515        command.Parameters.AddWithValue("@correlation_id", correlationId);
 6339516        var sinceParameter = command.Parameters.Add("@since", SqlDbType.DateTime2);
 6339517        sinceParameter.Scale = 7;
 6339518        sinceParameter.Value = sinceUtc.UtcDateTime;
 6339519        command.Parameters.AddWithValue("@limit", batchSize);
 6339520        if (afterCreatedAtUtc is not null)
 521        {
 265522            var cursorParameter = command.Parameters.Add("@after_created_at", SqlDbType.DateTime2);
 265523            cursorParameter.Scale = 7;
 265524            cursorParameter.Value = afterCreatedAtUtc.Value.UtcDateTime;
 265525            command.Parameters.AddWithValue("@after_id", afterId ?? throw new ArgumentNullException(nameof(afterId)));
 526        }
 527
 6339528        return await ReadMessagesAsync(command, batchSize, cancellationToken).ConfigureAwait(false);
 6312529    }
 530
 531    /// <summary>
 532    /// The full rows (envelope included) for <paramref name="ids"/> under
 533    /// <paramref name="correlationId"/>, in sweep order — how the dispatch sweep hydrates the
 534    /// header-only acknowledged rows it still has to deliver. A row pruned between the sweep's
 535    /// page and this read is simply absent.
 536    /// </summary>
 537    public async Task<IReadOnlyList<SqlServerChannelMessage>> LoadMessagesByIdAsync(
 538        string correlationId,
 539        IReadOnlyList<Guid> ids,
 540        CancellationToken cancellationToken)
 541    {
 23542        if (ids.Count == 0)
 2543            return [];
 544
 21545        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 21546        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 19547        await using var command = connection.CreateCommand();
 548        // One parameter per id (the sweep hands over at most a page): a joined literal list
 549        // would put ids into SQL text, and SQL Server has no array parameter to bind instead.
 19550        var placeholders = new string[ids.Count];
 78551        for (var i = 0; i < ids.Count; i++)
 552        {
 20553            placeholders[i] = $"@id{i}";
 20554            command.Parameters.Add(placeholders[i], SqlDbType.UniqueIdentifier).Value = ids[i];
 555        }
 556
 19557        command.CommandText =
 19558            $"""
 19559            SELECT id, correlation_id, envelope_json, created_at, acked_at, acked_seq
 19560            FROM {MessageTable}
 19561            WHERE correlation_id = @correlation_id
 19562              AND id IN ({string.Join(", ", placeholders)})
 19563              AND expires_at > SYSUTCDATETIME()
 19564            ORDER BY created_at, id;
 19565            """;
 19566        command.Parameters.AddWithValue("@correlation_id", correlationId);
 19567        return await ReadMessagesAsync(command, ids.Count, cancellationToken).ConfigureAwait(false);
 20568    }
 569
 570    private static async Task<IReadOnlyList<SqlServerChannelMessage>> ReadMessagesAsync(SqlCommand command, int capacity
 571    {
 6358572        var messages = new List<SqlServerChannelMessage>(capacity);
 6358573        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 7190574        while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 847575            messages.Add(new SqlServerChannelMessage(
 847576                reader.GetGuid(0),
 847577                reader.GetString(1),
 847578                reader.IsDBNull(2) ? null : reader.GetString(2),
 847579                new DateTimeOffset(reader.GetDateTime(3), TimeSpan.Zero),
 847580                reader.IsDBNull(4) ? null : new DateTimeOffset(reader.GetDateTime(4), TimeSpan.Zero),
 847581                reader.IsDBNull(5) ? null : reader.GetInt64(5)));
 6330582        return messages;
 6330583    }
 584
 585    /// <summary>
 586    /// Atomically claims a message for live delivery: sets <c>acked_at</c> unless the publisher has
 587    /// already routed it to the lost-subscriber path (<c>recovery_claimed</c>). Returns <c>false</c>
 588    /// when recovery owns the message, so a slow-but-live waiter does not deliver a response the
 589    /// recovery callback already handled. Multiple processes may each win this claim, preserving
 590    /// cross-process fan-out, because it gates only on <c>recovery_claimed</c>, not on <c>acked_at</c>.
 591    /// </summary>
 592    public async Task<bool> TryClaimForDeliveryAsync(Guid messageId, CancellationToken cancellationToken)
 593    {
 569594        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 563595        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 563596        await using var command = connection.CreateCommand();
 597        // NEXT VALUE FOR is not allowed inside CASE/COALESCE, so the sequence value is drawn into
 598        // a variable first — one batch, one round trip; the unused draw on an already-acked row
 599        // just leaves a harmless sequence gap. The sequence is stamped ONLY when this same update
 600        // transitions acked_at from null (SET expressions read the pre-update row): a row acked by
 601        // a pre-sequence build must stay permanently unsequenced — back-filling it on a later
 602        // fan-out re-claim would pair an OLD acked_at with a FRESH sequence value, and a waiter
 603        // that registered in the original ack's tick would then read the tie as post-registration
 604        // fan-out, replaying a response its predecessor consumed.
 563605        command.CommandText =
 563606            $"""
 563607            DECLARE @seq bigint = NEXT VALUE FOR {AckSequence};
 563608            UPDATE {MessageTable}
 563609            SET acked_at = COALESCE(acked_at, SYSUTCDATETIME()),
 563610                acked_seq = CASE WHEN acked_at IS NULL THEN @seq ELSE acked_seq END
 563611            OUTPUT inserted.id
 563612            WHERE id = @id AND recovery_claimed = 0 AND expires_at > SYSUTCDATETIME();
 563613            """;
 563614        command.Parameters.AddWithValue("@id", messageId);
 563615        var result = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
 563616        return result is not null and not DBNull;
 563617    }
 618
 619    /// <summary>
 620    /// Atomically claims a message for the lost-subscriber path: sets <c>recovery_claimed</c> only
 621    /// while no waiter has delivered (<c>acked_at IS NULL</c>). Returns <c>true</c> when recovery wins;
 622    /// <c>false</c> means a live waiter already took the message, so the publisher must not also fire
 623    /// the recovery callback. Row-level locking serializes this against <see cref="TryClaimForDeliveryAsync"/>.
 624    /// </summary>
 625    public async Task<bool> TryClaimForRecoveryAsync(Guid messageId, CancellationToken cancellationToken)
 626    {
 7627        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 7628        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 7629        await using var command = connection.CreateCommand();
 7630        command.CommandText =
 7631            $"""
 7632            UPDATE {MessageTable}
 7633            SET recovery_claimed = 1
 7634            OUTPUT inserted.id
 7635            WHERE id = @id AND acked_at IS NULL;
 7636            """;
 7637        command.Parameters.AddWithValue("@id", messageId);
 7638        var result = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
 7639        return result is not null and not DBNull;
 7640    }
 641
 642    /// <summary>
 643    /// One round trip for a subscription's registration watermark: the server's UTC clock (for
 644    /// the created-at bound) and a fresh position in the monotonic ack sequence (for the exact
 645    /// acked-history bound — see the watermark in the shared channel base).
 646    /// </summary>
 647    public async Task<(DateTimeOffset ServerTimeUtc, long StartSeq)> GetSubscriptionStartAsync(CancellationToken cancell
 648    {
 462649        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 460650        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 460651        await using var command = connection.CreateCommand();
 460652        command.CommandText = $"SELECT SYSUTCDATETIME(), NEXT VALUE FOR {AckSequence};";
 460653        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 460654        await reader.ReadAsync(cancellationToken).ConfigureAwait(false);
 460655        return (new DateTimeOffset(reader.GetDateTime(0), TimeSpan.Zero), reader.GetInt64(1));
 460656    }
 657
 658    /// <summary>Returns the database server's current UTC time, used as a clock-safe delivery watermark.</summary>
 659    public async Task<DateTimeOffset> GetServerTimeUtcAsync(CancellationToken cancellationToken)
 660    {
 5661        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 5662        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 3663        await using var command = connection.CreateCommand();
 3664        command.CommandText = "SELECT SYSUTCDATETIME();";
 3665        var result = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
 3666        return result switch
 3667        {
 0668            DateTimeOffset dto => dto.ToUniversalTime(),
 3669            DateTime dt => new DateTimeOffset(DateTime.SpecifyKind(dt, DateTimeKind.Utc), TimeSpan.Zero),
 0670            _ => DateTimeOffset.UtcNow
 3671        };
 3672    }
 673
 674    public async Task<bool> IsMessageAcknowledgedAsync(Guid messageId, CancellationToken cancellationToken)
 675    {
 138676        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 136677        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 134678        await using var command = connection.CreateCommand();
 134679        command.CommandText =
 134680            $"""
 134681            SELECT CAST(CASE WHEN acked_at IS NOT NULL THEN 1 ELSE 0 END AS bit)
 134682            FROM {MessageTable}
 134683            WHERE id = @id AND expires_at > SYSUTCDATETIME();
 134684            """;
 134685        command.Parameters.AddWithValue("@id", messageId);
 134686        var result = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
 134687        return result is bool acknowledged && acknowledged;
 134688    }
 689
 690    public async Task UpsertSubscriberAsync(string correlationId, Guid registrationId, string instanceId, TimeSpan ttl, 
 691    {
 464692        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 464693        if (ShouldPrune(ref _lastSubscriberPruneTicks))
 464694            await PruneExpiredSubscribersAsync(correlationId, cancellationToken).ConfigureAwait(false);
 695
 464696        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 464697        await using var command = connection.CreateCommand();
 464698        command.CommandText =
 464699            $"""
 464700            MERGE {SubscriberTable} WITH (HOLDLOCK) AS target
 464701            USING (SELECT @correlation_id AS correlation_id, @registration_id AS registration_id) AS source
 464702                ON target.correlation_id = source.correlation_id AND target.registration_id = source.registration_id
 464703            WHEN MATCHED THEN
 464704                UPDATE SET instance_id = @instance_id,
 464705                           expires_at = {AddMilliseconds("@ttl_ms")}
 464706            WHEN NOT MATCHED THEN
 464707                INSERT (correlation_id, registration_id, instance_id, expires_at)
 464708                VALUES (@correlation_id, @registration_id, @instance_id, {AddMilliseconds("@ttl_ms")});
 464709            """;
 464710        command.Parameters.AddWithValue("@correlation_id", correlationId);
 464711        command.Parameters.AddWithValue("@registration_id", registrationId);
 464712        command.Parameters.AddWithValue("@instance_id", instanceId);
 464713        command.Parameters.AddWithValue("@ttl_ms", (long)ttl.TotalMilliseconds);
 464714        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 464715    }
 716
 717    public async Task HeartbeatSubscribersAsync(
 718        string instanceId,
 719        IReadOnlyList<(string CorrelationId, Guid RegistrationId)> registrations,
 720        TimeSpan ttl,
 721        CancellationToken cancellationToken)
 722    {
 1911723        if (registrations.Count == 0)
 2724            return;
 725
 1909726        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 1901727        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 728
 729        // Two parameters per row plus instance/ttl stays under SQL Server's 2100-parameter cap.
 730        const int batchSize = 1000;
 7592731        for (var offset = 0; offset < registrations.Count; offset += batchSize)
 732        {
 1899733            var count = Math.Min(batchSize, registrations.Count - offset);
 1899734            await using var command = connection.CreateCommand();
 1899735            var sourceRows = new string[count];
 8878736            for (var index = 0; index < count; index++)
 737            {
 2540738                var (correlationId, registrationId) = registrations[offset + index];
 2540739                sourceRows[index] = $"(@correlation_id_{index}, @registration_id_{index})";
 2540740                command.Parameters.AddWithValue($"@correlation_id_{index}", correlationId);
 2540741                command.Parameters.AddWithValue($"@registration_id_{index}", registrationId);
 742            }
 743
 744            // MERGE upsert rather than a bare UPDATE, in the same WITH (HOLDLOCK) style as
 745            // UpsertSubscriberAsync: the caller only heartbeats registrations that are live in this
 746            // process, so a missing row means the pruner deleted it (e.g. after a >timeout stall)
 747            // — re-creating it here is what brings the waiter back from "permanently invisible".
 1899748            command.CommandText =
 1899749                $"""
 1899750                MERGE {SubscriberTable} WITH (HOLDLOCK) AS target
 1899751                USING (VALUES {string.Join(", ", sourceRows)}) AS source (correlation_id, registration_id)
 1899752                    ON target.correlation_id = source.correlation_id AND target.registration_id = source.registration_id
 1899753                WHEN MATCHED THEN
 1899754                    UPDATE SET instance_id = @instance_id,
 1899755                               expires_at = {AddMilliseconds("@ttl_ms")}
 1899756                WHEN NOT MATCHED THEN
 1899757                    INSERT (correlation_id, registration_id, instance_id, expires_at)
 1899758                    VALUES (source.correlation_id, source.registration_id, @instance_id, {AddMilliseconds("@ttl_ms")});
 1899759                """;
 1899760            command.Parameters.AddWithValue("@instance_id", instanceId);
 1899761            command.Parameters.AddWithValue("@ttl_ms", (long)ttl.TotalMilliseconds);
 1899762            await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 1897763        }
 1899764    }
 765
 766    public async Task DeleteSubscriberAsync(string correlationId, Guid registrationId, CancellationToken cancellationTok
 767    {
 505768        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 495769        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 492770        await using var command = connection.CreateCommand();
 492771        command.CommandText = $"DELETE FROM {SubscriberTable} WHERE correlation_id = @correlation_id AND registration_id
 492772        command.Parameters.AddWithValue("@correlation_id", correlationId);
 492773        command.Parameters.AddWithValue("@registration_id", registrationId);
 492774        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 492775    }
 776
 777    public async Task<long> CountActiveSubscribersAsync(string correlationId, CancellationToken cancellationToken)
 778    {
 623779        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 613780        if (ShouldPrune(ref _lastSubscriberPruneTicks))
 613781            await PruneExpiredSubscribersAsync(correlationId, cancellationToken).ConfigureAwait(false);
 782
 611783        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 611784        await using var command = connection.CreateCommand();
 611785        command.CommandText =
 611786            $"""
 611787            SELECT COUNT_BIG(*)
 611788            FROM {SubscriberTable}
 611789            WHERE correlation_id = @correlation_id AND expires_at > SYSUTCDATETIME();
 611790            """;
 611791        command.Parameters.AddWithValue("@correlation_id", correlationId);
 611792        var result = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
 611793        return result is long count ? count : 0L;
 611794    }
 795
 796    /// <summary>
 797    /// Bound on the table-wide prunes (durable-flow-store parity). They run inline on the publish
 798    /// and probe paths, and an unbounded DELETE over a backlog past SQL Server's ~5,000-lock
 799    /// escalation threshold takes a table lock that stalls concurrent delivery claims on the same
 800    /// table — long enough for a live waiter's claim to lose to the recovery claim. A bounded
 801    /// batch drains a backlog across successive calls instead.
 802    /// </summary>
 803    private const int PruneBatchSize = 1000;
 804
 805    /// <summary>The bounded table-wide prune statement for <paramref name="table"/>.</summary>
 806    internal static string ExpiredPruneSql(string table)
 647807        => $"DELETE TOP ({PruneBatchSize}) FROM {table} WHERE expires_at <= SYSUTCDATETIME();";
 808
 809    private async Task PruneExpiredRecoveryAsync(string? correlationId, CancellationToken cancellationToken)
 810    {
 40811        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 38812        await using var command = connection.CreateCommand();
 38813        command.CommandText = correlationId is null
 38814            ? ExpiredPruneSql(RecoveryTable)
 38815            : $"DELETE FROM {RecoveryTable} WHERE correlation_id = @correlation_id AND expires_at <= SYSUTCDATETIME();";
 38816        if (correlationId is not null)
 37817            command.Parameters.AddWithValue("@correlation_id", correlationId);
 38818        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 38819    }
 820
 821    private async Task PruneExpiredMessagesAsync(CancellationToken cancellationToken)
 822    {
 646823        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 644824        await using var command = connection.CreateCommand();
 644825        command.CommandText = ExpiredPruneSql(MessageTable);
 644826        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 644827    }
 828
 829    private async Task PruneExpiredSubscribersAsync(string? correlationId, CancellationToken cancellationToken)
 830    {
 1079831        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1075832        await using var command = connection.CreateCommand();
 1075833        command.CommandText = correlationId is null
 1075834            ? ExpiredPruneSql(SubscriberTable)
 1075835            : $"DELETE FROM {SubscriberTable} WHERE correlation_id = @correlation_id AND expires_at <= SYSUTCDATETIME();
 1075836        if (correlationId is not null)
 1075837            command.Parameters.AddWithValue("@correlation_id", correlationId);
 1075838        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 1075839    }
 840
 841    private async Task<SqlConnection> OpenConnectionAsync(CancellationToken cancellationToken)
 842    {
 14840843        var connection = new SqlConnection(_connectionString);
 844        try
 845        {
 14840846            await connection.OpenAsync(cancellationToken).ConfigureAwait(false);
 14771847            return connection;
 848        }
 69849        catch
 850        {
 69851            await connection.DisposeAsync().ConfigureAwait(false);
 69852            throw;
 853        }
 14771854    }
 855
 856    public static void ValidateIdentifier(string? value, string name)
 857    {
 3686858        if (string.IsNullOrWhiteSpace(value))
 4859            throw new InvalidOperationException($"{nameof(SqlServerAsyncResponseChannelOptions)}.{name} must be configur
 3682860        if (!IsIdentifier(value))
 6861            throw new InvalidOperationException(
 6862                $"{nameof(SqlServerAsyncResponseChannelOptions)}.{name} '{value}' must be a simple SQL Server identifier
 863        // sysname caps identifiers at 128; an over-limit name fails at DDL time with a raw
 864        // "identifier too long" error instead of an actionable configuration error.
 3676865        if (value.Length > IdentifierCap)
 2866            throw new InvalidOperationException(
 2867                $"{nameof(SqlServerAsyncResponseChannelOptions)}.{name} '{value}' is {value.Length} characters; SQL Serv
 3674868    }
 869
 870    private static bool IsIdentifier(string value)
 871    {
 3690872        if (value.Length == 0 || !(char.IsAsciiLetter(value[0]) || value[0] == '_'))
 8873            return false;
 874
 206728875        foreach (var c in value)
 876        {
 99684877            if (!(char.IsAsciiLetterOrDigit(c) || c == '_'))
 4878                return false;
 879        }
 880
 3678881        return true;
 882    }
 883
 3773884    private static string Quote(string identifier) => "[" + identifier + "]";
 885
 886    /// <summary>SQL Server's identifier length cap (sysname); longer names error at DDL time.</summary>
 887    internal const int IdentifierCap = 128;
 888
 889    // Suffix space is RESERVED before capping in BOTH derived-name helpers: truncating the whole
 890    // "{table}{suffix}" let a maximum-length table name derive exactly its own name (the sequence
 891    // collided with the table in the schema-object namespace and CREATE SEQUENCE failed) or let
 892    // the table's two indexes derive one shared name (the second IF NOT EXISTS guard matched the
 893    // first index and silently skipped creation).
 894    private static string SequenceName(string table)
 1369895        => RelationalNamePlan.DerivedName(table, "_ack_seq", IdentifierCap);
 896
 897    private static string IndexName(string table, string suffix)
 4610898        => RelationalNamePlan.DerivedName(table, $"_{suffix}_idx", IdentifierCap);
 899
 900    /// <summary>
 901    /// Validates the effective schema-object name plan: the three configured tables plus the
 902    /// derived ack sequence must be pairwise distinct (they share SQL Server's schema-scoped
 903    /// object namespace, and a table whose name ends exactly where the reserved "_ack_seq" stem
 904    /// truncates derives its own name). Index names live in per-table namespaces and carry
 905    /// distinct reserved suffixes, so they cannot collide once the tables are distinct.
 906    /// Comparison is case-insensitive to match SQL Server's default catalog collations.
 907    /// </summary>
 908    public static void ValidateNamePlan(SqlServerAsyncResponseChannelOptions options)
 909    {
 920910        (string Role, string Name)[] plan =
 920911        [
 920912            ($"{nameof(options.RecoveryStateTable)} table", options.RecoveryStateTable),
 920913            ($"{nameof(options.MessageTable)} table", options.MessageTable),
 920914            ($"{nameof(options.SubscriberTable)} table", options.SubscriberTable),
 920915            ("ack sequence (derived from MessageTable)", SequenceName(options.MessageTable)),
 920916        ];
 920917        RelationalNamePlan.RequireDistinct(
 920918            plan,
 920919            nameof(SqlServerAsyncResponseChannelOptions),
 920920            ". Tables and the sequence derived from MessageTable share one schema-object namespace and must be distinct 
 920921            "(long names reserve suffix space by truncating the table stem). Shorten or de-overlap the configured table 
 914922    }
 923
 924    /// <summary>
 925    /// SQL expression adding a millisecond bigint parameter to the database clock. DATEADD only takes
 926    /// int arguments, so the value is split into whole seconds and a sub-second remainder — TTLs and
 927    /// retentions stay on the database clock, immune to app-side clock skew, without overflowing on
 928    /// long spans such as the 7-day recovery expiry.
 929    /// </summary>
 930    internal static string AddMilliseconds(string parameterName)
 6318931        => $"DATEADD(SECOND, CAST({parameterName} / 1000 AS int), DATEADD(MILLISECOND, CAST({parameterName} % 1000 AS in
 932
 86933    internal static bool IsTransient(Exception exception) => SqlServerTransientFaults.IsTransient(exception);
 934
 935    /// <summary>
 936    /// Stable application-lock resource for serializing schema creation. It must be deterministic
 937    /// across processes and identical to the transport store's resource for the same schema so both
 938    /// serialize their shared CREATE SCHEMA.
 939    /// </summary>
 940    internal static string SchemaLockResource(string schemaName)
 393941        => $"asyncresponse:ddl:{schemaName}";
 942
 943    /// <summary>
 944    /// Time-gates opportunistic pruning so the housekeeping DELETE runs at most once per
 945    /// <see cref="SqlServerAsyncResponseChannelOptions.PruneInterval"/> instead of on every operation.
 946    /// Read queries already filter on <c>expires_at</c>, so throttling pruning never affects correctness.
 947    /// </summary>
 948    private bool ShouldPrune(ref long lastTicks)
 949    {
 1764950        var interval = _options.PruneInterval;
 1764951        if (interval <= TimeSpan.Zero)
 1758952            return true;
 953
 6954        var now = DateTime.UtcNow.Ticks;
 6955        var last = Interlocked.Read(ref lastTicks);
 6956        return now - last >= interval.Ticks
 6957            && Interlocked.CompareExchange(ref lastTicks, now, last) == last;
 958    }
 959}
Generated by: ReportGenerator 5.5.11.0
09/22/2026 - 09:41:04
GitHub | reportgenerator.io

Methods/Properties

.ctor(Microsoft.Extensions.Options.IOptions`1<AsyncResponse.Channels.SqlServer.SqlServerAsyncResponseChannelOptions>)
get_Schema()
get_RecoveryTable()
get_MessageTable()
get_SubscriberTable()
get_AckSequence()
get_AckSequenceName()
EnsureCreatedAsync()
VerifyRelationsAsync(Microsoft.Data.SqlClient.SqlConnection,Microsoft.Data.SqlClient.SqlTransaction,System.Threading.CancellationToken)
ExpectedObjects()
ValidateManagedSchemaAsync()
SaveRecoveryStateAsync()
LoadRecoveryStatesAsync()
DeleteRecoveryStateAsync()
ScanRecoveryStateJsonAsync()
InsertMessageAsync(System.Guid,System.String,System.String,System.TimeSpan,System.Threading.CancellationToken)
InsertMessageOnceAsync()
LoadMessagesAsync()
LoadMessagesByIdAsync()
ReadMessagesAsync()
TryClaimForDeliveryAsync()
TryClaimForRecoveryAsync()
GetSubscriptionStartAsync()
GetServerTimeUtcAsync()
IsMessageAcknowledgedAsync()
UpsertSubscriberAsync()
HeartbeatSubscribersAsync()
DeleteSubscriberAsync()
CountActiveSubscribersAsync()
ExpiredPruneSql(System.String)
PruneExpiredRecoveryAsync()
PruneExpiredMessagesAsync()
PruneExpiredSubscribersAsync()
OpenConnectionAsync()
ValidateIdentifier(System.String,System.String)
IsIdentifier(System.String)
Quote(System.String)
SequenceName(System.String)
IndexName(System.String,System.String)
ValidateNamePlan(AsyncResponse.Channels.SqlServer.SqlServerAsyncResponseChannelOptions)
AddMilliseconds(System.String)
IsTransient(System.Exception)
SchemaLockResource(System.String)
ShouldPrune(System.Int64&)