| | | 1 | | using AsyncResponse.Internal; |
| | | 2 | | using Microsoft.Data.SqlClient; |
| | | 3 | | using System.Data; |
| | | 4 | | |
| | | 5 | | namespace 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> |
| | | 14 | | internal readonly record struct SqlServerChannelMessage( |
| | 3380 | 15 | | Guid Id, |
| | 1868 | 16 | | string CorrelationId, |
| | 762 | 17 | | string? EnvelopeJson, |
| | 2154 | 18 | | DateTimeOffset CreatedAtUtc, |
| | 1810 | 19 | | DateTimeOffset? AckedAtUtc = null, |
| | 14 | 20 | | long? AckedSeq = null); |
| | | 21 | | |
| | | 22 | | /// <summary>SQL helper for the SQL Server channel tables.</summary> |
| | | 23 | | internal 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; |
| | | 32 | | private readonly SemaphoreSlim _ensureGate = new(1, 1); |
| | | 33 | | private bool _created; |
| | | 34 | | private long _lastRecoveryPruneTicks; |
| | | 35 | | private long _lastMessagePruneTicks; |
| | | 36 | | private long _lastSubscriberPruneTicks; |
| | | 37 | | |
| | | 38 | | public SqlServerChannelSql(Microsoft.Extensions.Options.IOptions<SqlServerAsyncResponseChannelOptions> options) |
| | | 39 | | { |
| | | 40 | | _options = options.Value; |
| | | 41 | | _options.Validate(); |
| | | 42 | | _connectionString = _options.ConnectionString!; |
| | | 43 | | |
| | | 44 | | Schema = Quote(_options.SchemaName); |
| | | 45 | | RecoveryTable = $"{Schema}.{Quote(_options.RecoveryStateTable)}"; |
| | | 46 | | MessageTable = $"{Schema}.{Quote(_options.MessageTable)}"; |
| | | 47 | | SubscriberTable = $"{Schema}.{Quote(_options.SubscriberTable)}"; |
| | | 48 | | AckSequenceName = SequenceName(_options.MessageTable); |
| | | 49 | | AckSequence = $"{Schema}.{Quote(AckSequenceName)}"; |
| | | 50 | | } |
| | | 51 | | |
| | | 52 | | public string Schema { get; } |
| | | 53 | | public string RecoveryTable { get; } |
| | | 54 | | public string MessageTable { get; } |
| | | 55 | | 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> |
| | | 62 | | public string AckSequence { get; } |
| | | 63 | | |
| | | 64 | | private string AckSequenceName { get; } |
| | | 65 | | |
| | | 66 | | public async Task EnsureCreatedAsync(CancellationToken cancellationToken = default) |
| | | 67 | | { |
| | | 68 | | if (_created) |
| | | 69 | | return; |
| | | 70 | | |
| | | 71 | | 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. |
| | | 78 | | await ValidateManagedSchemaAsync(cancellationToken).ConfigureAwait(false); |
| | | 79 | | return; |
| | | 80 | | } |
| | | 81 | | |
| | | 82 | | await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false); |
| | | 83 | | try |
| | | 84 | | { |
| | | 85 | | if (_created) |
| | | 86 | | return; |
| | | 87 | | |
| | | 88 | | await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false); |
| | | 89 | | 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. |
| | | 96 | | await using (var lockCommand = connection.CreateCommand()) |
| | | 97 | | { |
| | | 98 | | lockCommand.Transaction = transaction; |
| | | 99 | | lockCommand.CommandText = |
| | | 100 | | """ |
| | | 101 | | DECLARE @lock_result int; |
| | | 102 | | EXEC @lock_result = sp_getapplock |
| | | 103 | | @Resource = @lock_resource, |
| | | 104 | | @LockMode = 'Exclusive', |
| | | 105 | | @LockOwner = 'Transaction', |
| | | 106 | | @LockTimeout = 60000; |
| | | 107 | | IF @lock_result < 0 |
| | | 108 | | THROW 51000, N'Failed to acquire the AsyncResponse DDL application lock.', 1; |
| | | 109 | | """; |
| | | 110 | | lockCommand.Parameters.AddWithValue("@lock_resource", SchemaLockResource(_options.SchemaName)); |
| | | 111 | | await lockCommand.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false); |
| | | 112 | | } |
| | | 113 | | |
| | | 114 | | await using var command = connection.CreateCommand(); |
| | | 115 | | command.Transaction = transaction; |
| | | 116 | | command.CommandText = |
| | | 117 | | $""" |
| | | 118 | | IF SCHEMA_ID(N'{_options.SchemaName}') IS NULL |
| | | 119 | | EXEC(N'CREATE SCHEMA {Schema}'); |
| | | 120 | | |
| | | 121 | | IF OBJECT_ID(N'{RecoveryTable}', N'U') IS NULL |
| | | 122 | | CREATE TABLE {RecoveryTable} ( |
| | | 123 | | correlation_id nvarchar(400) COLLATE Latin1_General_100_BIN2 NOT NULL, |
| | | 124 | | registration_id uniqueidentifier NOT NULL, |
| | | 125 | | state_json nvarchar(max) NOT NULL, |
| | | 126 | | expires_at datetime2 NOT NULL, |
| | | 127 | | registered_at datetime2 NOT NULL DEFAULT SYSUTCDATETIME(), |
| | | 128 | | PRIMARY KEY (correlation_id, registration_id) |
| | | 129 | | ); |
| | | 130 | | IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name = N'{IndexName(_options.RecoveryStateTable, "expires |
| | | 131 | | CREATE INDEX {Quote(IndexName(_options.RecoveryStateTable, "expires"))} |
| | | 132 | | ON {RecoveryTable} (expires_at); |
| | | 133 | | |
| | | 134 | | IF OBJECT_ID(N'{MessageTable}', N'U') IS NULL |
| | | 135 | | CREATE TABLE {MessageTable} ( |
| | | 136 | | id uniqueidentifier NOT NULL PRIMARY KEY NONCLUSTERED, |
| | | 137 | | correlation_id nvarchar(400) COLLATE Latin1_General_100_BIN2 NOT NULL, |
| | | 138 | | envelope_json nvarchar(max) NOT NULL, |
| | | 139 | | created_at datetime2 NOT NULL DEFAULT SYSUTCDATETIME(), |
| | | 140 | | expires_at datetime2 NOT NULL, |
| | | 141 | | acked_at datetime2 NULL, |
| | | 142 | | acked_seq bigint NULL, |
| | | 143 | | recovery_claimed bit NOT NULL DEFAULT 0 |
| | | 144 | | ); |
| | | 145 | | IF COL_LENGTH(N'{MessageTable}', N'acked_seq') IS NULL |
| | | 146 | | ALTER TABLE {MessageTable} ADD acked_seq bigint NULL; |
| | | 147 | | IF NOT EXISTS (SELECT 1 FROM sys.sequences WHERE name = N'{AckSequenceName}' AND schema_id = SCHEMA_ID(N |
| | | 148 | | CREATE SEQUENCE {AckSequence} AS bigint START WITH 1; |
| | | 149 | | IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name = N'{IndexName(_options.MessageTable, "correlation_c |
| | | 150 | | CREATE INDEX {Quote(IndexName(_options.MessageTable, "correlation_created"))} |
| | | 151 | | ON {MessageTable} (correlation_id, created_at); |
| | | 152 | | IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name = N'{IndexName(_options.MessageTable, "expires")}' A |
| | | 153 | | CREATE INDEX {Quote(IndexName(_options.MessageTable, "expires"))} |
| | | 154 | | ON {MessageTable} (expires_at); |
| | | 155 | | |
| | | 156 | | IF OBJECT_ID(N'{SubscriberTable}', N'U') IS NULL |
| | | 157 | | CREATE TABLE {SubscriberTable} ( |
| | | 158 | | correlation_id nvarchar(400) COLLATE Latin1_General_100_BIN2 NOT NULL, |
| | | 159 | | registration_id uniqueidentifier NOT NULL, |
| | | 160 | | instance_id nvarchar(200) NOT NULL, |
| | | 161 | | expires_at datetime2 NOT NULL, |
| | | 162 | | PRIMARY KEY (correlation_id, registration_id) |
| | | 163 | | ); |
| | | 164 | | IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name = N'{IndexName(_options.SubscriberTable, "expires")} |
| | | 165 | | CREATE INDEX {Quote(IndexName(_options.SubscriberTable, "expires"))} |
| | | 166 | | ON {SubscriberTable} (expires_at); |
| | | 167 | | """; |
| | | 168 | | try |
| | | 169 | | { |
| | | 170 | | await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false); |
| | | 171 | | } |
| | | 172 | | 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. |
| | | 181 | | await SqlServerRelationVerifier.ThrowDiagnosedCollisionAsync( |
| | | 182 | | OpenConnectionAsync, |
| | | 183 | | ex, |
| | | 184 | | _options.SchemaName, |
| | | 185 | | "channel", |
| | | 186 | | ExpectedObjects(), |
| | | 187 | | cancellationToken).ConfigureAwait(false); |
| | | 188 | | throw; |
| | | 189 | | } |
| | | 190 | | |
| | | 191 | | 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. |
| | | 200 | | await VerifyRelationsAsync(connection, transaction: null, cancellationToken).ConfigureAwait(false); |
| | | 201 | | _created = true; |
| | | 202 | | } |
| | | 203 | | finally |
| | | 204 | | { |
| | | 205 | | _ensureGate.Release(); |
| | | 206 | | } |
| | | 207 | | } |
| | | 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 |
| | | 217 | | => SqlServerRelationVerifier.VerifyAsync( |
| | | 218 | | connection, |
| | | 219 | | transaction, |
| | | 220 | | _options.SchemaName, |
| | | 221 | | "channel", |
| | | 222 | | ExpectedObjects(), |
| | | 223 | | 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() => |
| | | 231 | | [ |
| | | 232 | | new(_options.RecoveryStateTable, SqlServerObjectKind.Table, |
| | | 233 | | [ |
| | | 234 | | new("correlation_id", "nvarchar(400)", Nullable: false, RequiresBinaryCollation: true), |
| | | 235 | | new("registration_id", "uniqueidentifier", Nullable: false), |
| | | 236 | | new("state_json", "nvarchar(max)", Nullable: false), |
| | | 237 | | new("expires_at", "datetime2(7)", Nullable: false), |
| | | 238 | | new("registered_at", "datetime2(7)", Nullable: false, DefaultExpression: "(sysutcdatetime())") |
| | | 239 | | ], |
| | | 240 | | PrimaryKey: ["correlation_id", "registration_id"]), |
| | | 241 | | new(_options.MessageTable, SqlServerObjectKind.Table, |
| | | 242 | | [ |
| | | 243 | | new("id", "uniqueidentifier", Nullable: false), |
| | | 244 | | new("correlation_id", "nvarchar(400)", Nullable: false, RequiresBinaryCollation: true), |
| | | 245 | | new("envelope_json", "nvarchar(max)", Nullable: false), |
| | | 246 | | new("created_at", "datetime2(7)", Nullable: false, DefaultExpression: "(sysutcdatetime())"), |
| | | 247 | | new("expires_at", "datetime2(7)", Nullable: false), |
| | | 248 | | new("acked_at", "datetime2(7)", Nullable: true), |
| | | 249 | | new("acked_seq", "bigint", Nullable: true), |
| | | 250 | | new("recovery_claimed", "bit", Nullable: false, DefaultExpression: "((0))") |
| | | 251 | | ], |
| | | 252 | | PrimaryKey: ["id"]), |
| | | 253 | | new(_options.SubscriberTable, SqlServerObjectKind.Table, |
| | | 254 | | [ |
| | | 255 | | new("correlation_id", "nvarchar(400)", Nullable: false, RequiresBinaryCollation: true), |
| | | 256 | | new("registration_id", "uniqueidentifier", Nullable: false), |
| | | 257 | | new("instance_id", "nvarchar(200)", Nullable: false), |
| | | 258 | | new("expires_at", "datetime2(7)", Nullable: false) |
| | | 259 | | ], |
| | | 260 | | PrimaryKey: ["correlation_id", "registration_id"]), |
| | | 261 | | new(AckSequenceName, SqlServerObjectKind.Sequence), |
| | | 262 | | new(IndexName(_options.RecoveryStateTable, "expires"), SqlServerObjectKind.Index, |
| | | 263 | | OwningTable: _options.RecoveryStateTable, KeyColumns: ["expires_at"]), |
| | | 264 | | new(IndexName(_options.MessageTable, "correlation_created"), SqlServerObjectKind.Index, |
| | | 265 | | OwningTable: _options.MessageTable, KeyColumns: ["correlation_id", "created_at"]), |
| | | 266 | | new(IndexName(_options.MessageTable, "expires"), SqlServerObjectKind.Index, |
| | | 267 | | OwningTable: _options.MessageTable, KeyColumns: ["expires_at"]), |
| | | 268 | | new(IndexName(_options.SubscriberTable, "expires"), SqlServerObjectKind.Index, |
| | | 269 | | OwningTable: _options.SubscriberTable, KeyColumns: ["expires_at"]) |
| | | 270 | | ]; |
| | | 271 | | |
| | | 272 | | private async Task ValidateManagedSchemaAsync(CancellationToken cancellationToken) |
| | | 273 | | { |
| | | 274 | | await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false); |
| | | 275 | | try |
| | | 276 | | { |
| | | 277 | | if (_created) |
| | | 278 | | return; |
| | | 279 | | |
| | | 280 | | 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. |
| | | 285 | | await using (var command = connection.CreateCommand()) |
| | | 286 | | { |
| | | 287 | | command.CommandText = |
| | | 288 | | $""" |
| | | 289 | | SELECT |
| | | 290 | | CASE WHEN COL_LENGTH(N'{MessageTable}', N'acked_seq') IS NOT NULL THEN 1 ELSE 0 END, |
| | | 291 | | CASE WHEN EXISTS (SELECT 1 FROM sys.sequences WHERE name = N'{AckSequenceName}' AND schema_id = SC |
| | | 292 | | """; |
| | | 293 | | await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false); |
| | | 294 | | await reader.ReadAsync(cancellationToken).ConfigureAwait(false); |
| | | 295 | | hasColumn = reader.GetInt32(0) == 1; |
| | | 296 | | hasSequence = reader.GetInt32(1) == 1; |
| | | 297 | | } |
| | | 298 | | if (!hasColumn || !hasSequence) |
| | | 299 | | { |
| | | 300 | | throw new InvalidOperationException( |
| | | 301 | | $"The SQL Server channel schema is managed manually (AutoCreateSchema = false) but is missing " + |
| | | 302 | | $"objects this version requires: " + |
| | | 303 | | $"{(hasColumn ? "" : $"column {MessageTable}.acked_seq")}{(!hasColumn && !hasSequence ? " and " : "" |
| | | 304 | | $"Apply the migration and restart: " + |
| | | 305 | | $"IF COL_LENGTH(N'{MessageTable}', N'acked_seq') IS NULL ALTER TABLE {MessageTable} ADD acked_seq bi |
| | | 306 | | $"IF NOT EXISTS (SELECT 1 FROM sys.sequences WHERE name = N'{AckSequenceName}' AND schema_id = SCHEM |
| | | 307 | | "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. |
| | | 315 | | await VerifyRelationsAsync(connection, transaction: null, cancellationToken).ConfigureAwait(false); |
| | | 316 | | |
| | | 317 | | _created = true; |
| | | 318 | | } |
| | | 319 | | finally |
| | | 320 | | { |
| | | 321 | | _ensureGate.Release(); |
| | | 322 | | } |
| | | 323 | | } |
| | | 324 | | |
| | | 325 | | public async Task SaveRecoveryStateAsync(string correlationId, RecoveryState state, TimeSpan ttl, CancellationToken |
| | | 326 | | { |
| | | 327 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | | 328 | | await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false); |
| | | 329 | | 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. |
| | | 332 | | command.CommandText = |
| | | 333 | | $""" |
| | | 334 | | MERGE {RecoveryTable} WITH (HOLDLOCK) AS target |
| | | 335 | | USING (SELECT @correlation_id AS correlation_id, @registration_id AS registration_id) AS source |
| | | 336 | | ON target.correlation_id = source.correlation_id AND target.registration_id = source.registration_id |
| | | 337 | | WHEN MATCHED THEN |
| | | 338 | | UPDATE SET state_json = @state_json, |
| | | 339 | | expires_at = {AddMilliseconds("@ttl_ms")}, |
| | | 340 | | registered_at = SYSUTCDATETIME() |
| | | 341 | | WHEN NOT MATCHED THEN |
| | | 342 | | INSERT (correlation_id, registration_id, state_json, expires_at, registered_at) |
| | | 343 | | VALUES (@correlation_id, @registration_id, @state_json, {AddMilliseconds("@ttl_ms")}, SYSUTCDATETIME()); |
| | | 344 | | """; |
| | | 345 | | command.Parameters.AddWithValue("@correlation_id", correlationId); |
| | | 346 | | command.Parameters.AddWithValue("@registration_id", state.RegistrationId); |
| | | 347 | | command.Parameters.AddWithValue("@state_json", AsyncResponseJson.Serialize(state)); |
| | | 348 | | command.Parameters.AddWithValue("@ttl_ms", (long)ttl.TotalMilliseconds); |
| | | 349 | | await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false); |
| | | 350 | | } |
| | | 351 | | |
| | | 352 | | public async Task<IReadOnlyList<string>> LoadRecoveryStatesAsync(string correlationId, CancellationToken cancellatio |
| | | 353 | | { |
| | | 354 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | | 355 | | if (ShouldPrune(ref _lastRecoveryPruneTicks)) |
| | | 356 | | await PruneExpiredRecoveryAsync(correlationId, cancellationToken).ConfigureAwait(false); |
| | | 357 | | |
| | | 358 | | await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false); |
| | | 359 | | await using var command = connection.CreateCommand(); |
| | | 360 | | command.CommandText = |
| | | 361 | | $""" |
| | | 362 | | SELECT state_json |
| | | 363 | | FROM {RecoveryTable} |
| | | 364 | | WHERE correlation_id = @correlation_id AND expires_at > SYSUTCDATETIME() |
| | | 365 | | ORDER BY registered_at; |
| | | 366 | | """; |
| | | 367 | | command.Parameters.AddWithValue("@correlation_id", correlationId); |
| | | 368 | | |
| | | 369 | | var states = new List<string>(); |
| | | 370 | | await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false); |
| | | 371 | | while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false)) |
| | | 372 | | states.Add(reader.GetString(0)); |
| | | 373 | | return states; |
| | | 374 | | } |
| | | 375 | | |
| | | 376 | | public async Task<bool> DeleteRecoveryStateAsync(string correlationId, Guid registrationId, CancellationToken cancel |
| | | 377 | | { |
| | | 378 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | | 379 | | await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false); |
| | | 380 | | await using var command = connection.CreateCommand(); |
| | | 381 | | command.CommandText = $"DELETE FROM {RecoveryTable} WHERE correlation_id = @correlation_id AND registration_id = |
| | | 382 | | command.Parameters.AddWithValue("@correlation_id", correlationId); |
| | | 383 | | command.Parameters.AddWithValue("@registration_id", registrationId); |
| | | 384 | | return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0; |
| | | 385 | | } |
| | | 386 | | |
| | | 387 | | public async IAsyncEnumerable<string> ScanRecoveryStateJsonAsync([System.Runtime.CompilerServices.EnumeratorCancella |
| | | 388 | | { |
| | | 389 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | | 390 | | await PruneExpiredRecoveryAsync(null, cancellationToken).ConfigureAwait(false); |
| | | 391 | | |
| | | 392 | | await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false); |
| | | 393 | | await using var command = connection.CreateCommand(); |
| | | 394 | | command.CommandText = |
| | | 395 | | $""" |
| | | 396 | | SELECT state_json |
| | | 397 | | FROM {RecoveryTable} |
| | | 398 | | WHERE expires_at > SYSUTCDATETIME() |
| | | 399 | | ORDER BY registered_at; |
| | | 400 | | """; |
| | | 401 | | await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false); |
| | | 402 | | while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false)) |
| | | 403 | | yield return reader.GetString(0); |
| | | 404 | | } |
| | | 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 |
| | | 416 | | => AsyncResponseRetry.ExecuteAsync( |
| | | 417 | | token => InsertMessageOnceAsync(id, correlationId, envelopeJson, retention, token), |
| | | 418 | | IsTransient, |
| | | 419 | | _options.PublishMaxAttempts, |
| | | 420 | | _options.PublishRetryBaseDelay, |
| | | 421 | | _options.PublishRetryMaxDelay, |
| | | 422 | | cancellationToken); |
| | | 423 | | |
| | | 424 | | private async Task<SqlServerChannelMessage> InsertMessageOnceAsync(Guid id, string correlationId, string envelopeJso |
| | | 425 | | { |
| | | 426 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | | 427 | | if (ShouldPrune(ref _lastMessagePruneTicks)) |
| | | 428 | | await PruneExpiredMessagesAsync(cancellationToken).ConfigureAwait(false); |
| | | 429 | | |
| | | 430 | | await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false); |
| | | 431 | | await using var command = connection.CreateCommand(); |
| | | 432 | | command.CommandText = |
| | | 433 | | $""" |
| | | 434 | | INSERT INTO {MessageTable} (id, correlation_id, envelope_json, expires_at) |
| | | 435 | | OUTPUT inserted.created_at |
| | | 436 | | SELECT @id, @correlation_id, @envelope_json, {AddMilliseconds("@retention_ms")} |
| | | 437 | | WHERE NOT EXISTS (SELECT 1 FROM {MessageTable} WITH (UPDLOCK, HOLDLOCK) WHERE id = @id); |
| | | 438 | | """; |
| | | 439 | | command.Parameters.AddWithValue("@id", id); |
| | | 440 | | command.Parameters.AddWithValue("@correlation_id", correlationId); |
| | | 441 | | command.Parameters.AddWithValue("@envelope_json", envelopeJson); |
| | | 442 | | command.Parameters.AddWithValue("@retention_ms", (long)retention.TotalMilliseconds); |
| | | 443 | | |
| | | 444 | | object? createdAt = null; |
| | | 445 | | try |
| | | 446 | | { |
| | | 447 | | createdAt = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false); |
| | | 448 | | } |
| | | 449 | | catch (SqlException ex) when (ex.Number is PrimaryKeyViolation or UniqueIndexViolation) |
| | | 450 | | { |
| | | 451 | | } |
| | | 452 | | |
| | | 453 | | if (createdAt is DateTime insertedCreatedAt) |
| | | 454 | | 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. |
| | | 464 | | await using var lookup = connection.CreateCommand(); |
| | | 465 | | lookup.CommandText = $"SELECT created_at, acked_at, acked_seq FROM {MessageTable} WHERE id = @id;"; |
| | | 466 | | lookup.Parameters.AddWithValue("@id", id); |
| | | 467 | | await using var existing = await lookup.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false); |
| | | 468 | | if (await existing.ReadAsync(cancellationToken).ConfigureAwait(false)) |
| | | 469 | | { |
| | | 470 | | return new SqlServerChannelMessage( |
| | | 471 | | id, |
| | | 472 | | correlationId, |
| | | 473 | | envelopeJson, |
| | | 474 | | new DateTimeOffset(existing.GetDateTime(0), TimeSpan.Zero), |
| | | 475 | | existing.IsDBNull(1) ? null : new DateTimeOffset(existing.GetDateTime(1), TimeSpan.Zero), |
| | | 476 | | 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. |
| | | 483 | | throw new InvalidOperationException( |
| | | 484 | | $"SQL Server response insert for message {id} found no row after a duplicate: the original no longer exists |
| | | 485 | | } |
| | | 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 | | { |
| | | 495 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | | 496 | | await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false); |
| | | 497 | | 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. |
| | | 504 | | command.CommandText = |
| | | 505 | | $""" |
| | | 506 | | SELECT id, correlation_id, CASE WHEN acked_at IS NULL THEN envelope_json END, created_at, acked_at, acked_se |
| | | 507 | | FROM {MessageTable} |
| | | 508 | | WHERE correlation_id = @correlation_id |
| | | 509 | | AND created_at >= @since |
| | | 510 | | AND expires_at > SYSUTCDATETIME() |
| | | 511 | | {(afterCreatedAtUtc is null ? "" : "AND (created_at > @after_created_at OR (created_at = @after_created_at |
| | | 512 | | ORDER BY created_at, id |
| | | 513 | | OFFSET 0 ROWS FETCH NEXT @limit ROWS ONLY; |
| | | 514 | | """; |
| | | 515 | | command.Parameters.AddWithValue("@correlation_id", correlationId); |
| | | 516 | | var sinceParameter = command.Parameters.Add("@since", SqlDbType.DateTime2); |
| | | 517 | | sinceParameter.Scale = 7; |
| | | 518 | | sinceParameter.Value = sinceUtc.UtcDateTime; |
| | | 519 | | command.Parameters.AddWithValue("@limit", batchSize); |
| | | 520 | | if (afterCreatedAtUtc is not null) |
| | | 521 | | { |
| | | 522 | | var cursorParameter = command.Parameters.Add("@after_created_at", SqlDbType.DateTime2); |
| | | 523 | | cursorParameter.Scale = 7; |
| | | 524 | | cursorParameter.Value = afterCreatedAtUtc.Value.UtcDateTime; |
| | | 525 | | command.Parameters.AddWithValue("@after_id", afterId ?? throw new ArgumentNullException(nameof(afterId))); |
| | | 526 | | } |
| | | 527 | | |
| | | 528 | | return await ReadMessagesAsync(command, batchSize, cancellationToken).ConfigureAwait(false); |
| | | 529 | | } |
| | | 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 | | { |
| | | 542 | | if (ids.Count == 0) |
| | | 543 | | return []; |
| | | 544 | | |
| | | 545 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | | 546 | | await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false); |
| | | 547 | | 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. |
| | | 550 | | var placeholders = new string[ids.Count]; |
| | | 551 | | for (var i = 0; i < ids.Count; i++) |
| | | 552 | | { |
| | | 553 | | placeholders[i] = $"@id{i}"; |
| | | 554 | | command.Parameters.Add(placeholders[i], SqlDbType.UniqueIdentifier).Value = ids[i]; |
| | | 555 | | } |
| | | 556 | | |
| | | 557 | | command.CommandText = |
| | | 558 | | $""" |
| | | 559 | | SELECT id, correlation_id, envelope_json, created_at, acked_at, acked_seq |
| | | 560 | | FROM {MessageTable} |
| | | 561 | | WHERE correlation_id = @correlation_id |
| | | 562 | | AND id IN ({string.Join(", ", placeholders)}) |
| | | 563 | | AND expires_at > SYSUTCDATETIME() |
| | | 564 | | ORDER BY created_at, id; |
| | | 565 | | """; |
| | | 566 | | command.Parameters.AddWithValue("@correlation_id", correlationId); |
| | | 567 | | return await ReadMessagesAsync(command, ids.Count, cancellationToken).ConfigureAwait(false); |
| | | 568 | | } |
| | | 569 | | |
| | | 570 | | private static async Task<IReadOnlyList<SqlServerChannelMessage>> ReadMessagesAsync(SqlCommand command, int capacity |
| | | 571 | | { |
| | | 572 | | var messages = new List<SqlServerChannelMessage>(capacity); |
| | | 573 | | await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false); |
| | | 574 | | while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false)) |
| | | 575 | | messages.Add(new SqlServerChannelMessage( |
| | | 576 | | reader.GetGuid(0), |
| | | 577 | | reader.GetString(1), |
| | | 578 | | reader.IsDBNull(2) ? null : reader.GetString(2), |
| | | 579 | | new DateTimeOffset(reader.GetDateTime(3), TimeSpan.Zero), |
| | | 580 | | reader.IsDBNull(4) ? null : new DateTimeOffset(reader.GetDateTime(4), TimeSpan.Zero), |
| | | 581 | | reader.IsDBNull(5) ? null : reader.GetInt64(5))); |
| | | 582 | | return messages; |
| | | 583 | | } |
| | | 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 | | { |
| | | 594 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | | 595 | | await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false); |
| | | 596 | | 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. |
| | | 605 | | command.CommandText = |
| | | 606 | | $""" |
| | | 607 | | DECLARE @seq bigint = NEXT VALUE FOR {AckSequence}; |
| | | 608 | | UPDATE {MessageTable} |
| | | 609 | | SET acked_at = COALESCE(acked_at, SYSUTCDATETIME()), |
| | | 610 | | acked_seq = CASE WHEN acked_at IS NULL THEN @seq ELSE acked_seq END |
| | | 611 | | OUTPUT inserted.id |
| | | 612 | | WHERE id = @id AND recovery_claimed = 0 AND expires_at > SYSUTCDATETIME(); |
| | | 613 | | """; |
| | | 614 | | command.Parameters.AddWithValue("@id", messageId); |
| | | 615 | | var result = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false); |
| | | 616 | | return result is not null and not DBNull; |
| | | 617 | | } |
| | | 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 | | { |
| | | 627 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | | 628 | | await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false); |
| | | 629 | | await using var command = connection.CreateCommand(); |
| | | 630 | | command.CommandText = |
| | | 631 | | $""" |
| | | 632 | | UPDATE {MessageTable} |
| | | 633 | | SET recovery_claimed = 1 |
| | | 634 | | OUTPUT inserted.id |
| | | 635 | | WHERE id = @id AND acked_at IS NULL; |
| | | 636 | | """; |
| | | 637 | | command.Parameters.AddWithValue("@id", messageId); |
| | | 638 | | var result = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false); |
| | | 639 | | return result is not null and not DBNull; |
| | | 640 | | } |
| | | 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 | | { |
| | | 649 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | | 650 | | await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false); |
| | | 651 | | await using var command = connection.CreateCommand(); |
| | | 652 | | command.CommandText = $"SELECT SYSUTCDATETIME(), NEXT VALUE FOR {AckSequence};"; |
| | | 653 | | await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false); |
| | | 654 | | await reader.ReadAsync(cancellationToken).ConfigureAwait(false); |
| | | 655 | | return (new DateTimeOffset(reader.GetDateTime(0), TimeSpan.Zero), reader.GetInt64(1)); |
| | | 656 | | } |
| | | 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 | | { |
| | | 661 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | | 662 | | await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false); |
| | | 663 | | await using var command = connection.CreateCommand(); |
| | | 664 | | command.CommandText = "SELECT SYSUTCDATETIME();"; |
| | | 665 | | var result = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false); |
| | | 666 | | return result switch |
| | | 667 | | { |
| | | 668 | | DateTimeOffset dto => dto.ToUniversalTime(), |
| | | 669 | | DateTime dt => new DateTimeOffset(DateTime.SpecifyKind(dt, DateTimeKind.Utc), TimeSpan.Zero), |
| | | 670 | | _ => DateTimeOffset.UtcNow |
| | | 671 | | }; |
| | | 672 | | } |
| | | 673 | | |
| | | 674 | | public async Task<bool> IsMessageAcknowledgedAsync(Guid messageId, CancellationToken cancellationToken) |
| | | 675 | | { |
| | | 676 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | | 677 | | await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false); |
| | | 678 | | await using var command = connection.CreateCommand(); |
| | | 679 | | command.CommandText = |
| | | 680 | | $""" |
| | | 681 | | SELECT CAST(CASE WHEN acked_at IS NOT NULL THEN 1 ELSE 0 END AS bit) |
| | | 682 | | FROM {MessageTable} |
| | | 683 | | WHERE id = @id AND expires_at > SYSUTCDATETIME(); |
| | | 684 | | """; |
| | | 685 | | command.Parameters.AddWithValue("@id", messageId); |
| | | 686 | | var result = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false); |
| | | 687 | | return result is bool acknowledged && acknowledged; |
| | | 688 | | } |
| | | 689 | | |
| | | 690 | | public async Task UpsertSubscriberAsync(string correlationId, Guid registrationId, string instanceId, TimeSpan ttl, |
| | | 691 | | { |
| | | 692 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | | 693 | | if (ShouldPrune(ref _lastSubscriberPruneTicks)) |
| | | 694 | | await PruneExpiredSubscribersAsync(correlationId, cancellationToken).ConfigureAwait(false); |
| | | 695 | | |
| | | 696 | | await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false); |
| | | 697 | | await using var command = connection.CreateCommand(); |
| | | 698 | | command.CommandText = |
| | | 699 | | $""" |
| | | 700 | | MERGE {SubscriberTable} WITH (HOLDLOCK) AS target |
| | | 701 | | USING (SELECT @correlation_id AS correlation_id, @registration_id AS registration_id) AS source |
| | | 702 | | ON target.correlation_id = source.correlation_id AND target.registration_id = source.registration_id |
| | | 703 | | WHEN MATCHED THEN |
| | | 704 | | UPDATE SET instance_id = @instance_id, |
| | | 705 | | expires_at = {AddMilliseconds("@ttl_ms")} |
| | | 706 | | WHEN NOT MATCHED THEN |
| | | 707 | | INSERT (correlation_id, registration_id, instance_id, expires_at) |
| | | 708 | | VALUES (@correlation_id, @registration_id, @instance_id, {AddMilliseconds("@ttl_ms")}); |
| | | 709 | | """; |
| | | 710 | | command.Parameters.AddWithValue("@correlation_id", correlationId); |
| | | 711 | | command.Parameters.AddWithValue("@registration_id", registrationId); |
| | | 712 | | command.Parameters.AddWithValue("@instance_id", instanceId); |
| | | 713 | | command.Parameters.AddWithValue("@ttl_ms", (long)ttl.TotalMilliseconds); |
| | | 714 | | await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false); |
| | | 715 | | } |
| | | 716 | | |
| | | 717 | | public async Task HeartbeatSubscribersAsync( |
| | | 718 | | string instanceId, |
| | | 719 | | IReadOnlyList<(string CorrelationId, Guid RegistrationId)> registrations, |
| | | 720 | | TimeSpan ttl, |
| | | 721 | | CancellationToken cancellationToken) |
| | | 722 | | { |
| | | 723 | | if (registrations.Count == 0) |
| | | 724 | | return; |
| | | 725 | | |
| | | 726 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | | 727 | | 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; |
| | | 731 | | for (var offset = 0; offset < registrations.Count; offset += batchSize) |
| | | 732 | | { |
| | | 733 | | var count = Math.Min(batchSize, registrations.Count - offset); |
| | | 734 | | await using var command = connection.CreateCommand(); |
| | | 735 | | var sourceRows = new string[count]; |
| | | 736 | | for (var index = 0; index < count; index++) |
| | | 737 | | { |
| | | 738 | | var (correlationId, registrationId) = registrations[offset + index]; |
| | | 739 | | sourceRows[index] = $"(@correlation_id_{index}, @registration_id_{index})"; |
| | | 740 | | command.Parameters.AddWithValue($"@correlation_id_{index}", correlationId); |
| | | 741 | | 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". |
| | | 748 | | command.CommandText = |
| | | 749 | | $""" |
| | | 750 | | MERGE {SubscriberTable} WITH (HOLDLOCK) AS target |
| | | 751 | | USING (VALUES {string.Join(", ", sourceRows)}) AS source (correlation_id, registration_id) |
| | | 752 | | ON target.correlation_id = source.correlation_id AND target.registration_id = source.registration_id |
| | | 753 | | WHEN MATCHED THEN |
| | | 754 | | UPDATE SET instance_id = @instance_id, |
| | | 755 | | expires_at = {AddMilliseconds("@ttl_ms")} |
| | | 756 | | WHEN NOT MATCHED THEN |
| | | 757 | | INSERT (correlation_id, registration_id, instance_id, expires_at) |
| | | 758 | | VALUES (source.correlation_id, source.registration_id, @instance_id, {AddMilliseconds("@ttl_ms")}); |
| | | 759 | | """; |
| | | 760 | | command.Parameters.AddWithValue("@instance_id", instanceId); |
| | | 761 | | command.Parameters.AddWithValue("@ttl_ms", (long)ttl.TotalMilliseconds); |
| | | 762 | | await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false); |
| | | 763 | | } |
| | | 764 | | } |
| | | 765 | | |
| | | 766 | | public async Task DeleteSubscriberAsync(string correlationId, Guid registrationId, CancellationToken cancellationTok |
| | | 767 | | { |
| | | 768 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | | 769 | | await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false); |
| | | 770 | | await using var command = connection.CreateCommand(); |
| | | 771 | | command.CommandText = $"DELETE FROM {SubscriberTable} WHERE correlation_id = @correlation_id AND registration_id |
| | | 772 | | command.Parameters.AddWithValue("@correlation_id", correlationId); |
| | | 773 | | command.Parameters.AddWithValue("@registration_id", registrationId); |
| | | 774 | | await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false); |
| | | 775 | | } |
| | | 776 | | |
| | | 777 | | public async Task<long> CountActiveSubscribersAsync(string correlationId, CancellationToken cancellationToken) |
| | | 778 | | { |
| | | 779 | | await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); |
| | | 780 | | if (ShouldPrune(ref _lastSubscriberPruneTicks)) |
| | | 781 | | await PruneExpiredSubscribersAsync(correlationId, cancellationToken).ConfigureAwait(false); |
| | | 782 | | |
| | | 783 | | await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false); |
| | | 784 | | await using var command = connection.CreateCommand(); |
| | | 785 | | command.CommandText = |
| | | 786 | | $""" |
| | | 787 | | SELECT COUNT_BIG(*) |
| | | 788 | | FROM {SubscriberTable} |
| | | 789 | | WHERE correlation_id = @correlation_id AND expires_at > SYSUTCDATETIME(); |
| | | 790 | | """; |
| | | 791 | | command.Parameters.AddWithValue("@correlation_id", correlationId); |
| | | 792 | | var result = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false); |
| | | 793 | | return result is long count ? count : 0L; |
| | | 794 | | } |
| | | 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) |
| | | 807 | | => $"DELETE TOP ({PruneBatchSize}) FROM {table} WHERE expires_at <= SYSUTCDATETIME();"; |
| | | 808 | | |
| | | 809 | | private async Task PruneExpiredRecoveryAsync(string? correlationId, CancellationToken cancellationToken) |
| | | 810 | | { |
| | | 811 | | await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false); |
| | | 812 | | await using var command = connection.CreateCommand(); |
| | | 813 | | command.CommandText = correlationId is null |
| | | 814 | | ? ExpiredPruneSql(RecoveryTable) |
| | | 815 | | : $"DELETE FROM {RecoveryTable} WHERE correlation_id = @correlation_id AND expires_at <= SYSUTCDATETIME();"; |
| | | 816 | | if (correlationId is not null) |
| | | 817 | | command.Parameters.AddWithValue("@correlation_id", correlationId); |
| | | 818 | | await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false); |
| | | 819 | | } |
| | | 820 | | |
| | | 821 | | private async Task PruneExpiredMessagesAsync(CancellationToken cancellationToken) |
| | | 822 | | { |
| | | 823 | | await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false); |
| | | 824 | | await using var command = connection.CreateCommand(); |
| | | 825 | | command.CommandText = ExpiredPruneSql(MessageTable); |
| | | 826 | | await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false); |
| | | 827 | | } |
| | | 828 | | |
| | | 829 | | private async Task PruneExpiredSubscribersAsync(string? correlationId, CancellationToken cancellationToken) |
| | | 830 | | { |
| | | 831 | | await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false); |
| | | 832 | | await using var command = connection.CreateCommand(); |
| | | 833 | | command.CommandText = correlationId is null |
| | | 834 | | ? ExpiredPruneSql(SubscriberTable) |
| | | 835 | | : $"DELETE FROM {SubscriberTable} WHERE correlation_id = @correlation_id AND expires_at <= SYSUTCDATETIME(); |
| | | 836 | | if (correlationId is not null) |
| | | 837 | | command.Parameters.AddWithValue("@correlation_id", correlationId); |
| | | 838 | | await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false); |
| | | 839 | | } |
| | | 840 | | |
| | | 841 | | private async Task<SqlConnection> OpenConnectionAsync(CancellationToken cancellationToken) |
| | | 842 | | { |
| | | 843 | | var connection = new SqlConnection(_connectionString); |
| | | 844 | | try |
| | | 845 | | { |
| | | 846 | | await connection.OpenAsync(cancellationToken).ConfigureAwait(false); |
| | | 847 | | return connection; |
| | | 848 | | } |
| | | 849 | | catch |
| | | 850 | | { |
| | | 851 | | await connection.DisposeAsync().ConfigureAwait(false); |
| | | 852 | | throw; |
| | | 853 | | } |
| | | 854 | | } |
| | | 855 | | |
| | | 856 | | public static void ValidateIdentifier(string? value, string name) |
| | | 857 | | { |
| | | 858 | | if (string.IsNullOrWhiteSpace(value)) |
| | | 859 | | throw new InvalidOperationException($"{nameof(SqlServerAsyncResponseChannelOptions)}.{name} must be configur |
| | | 860 | | if (!IsIdentifier(value)) |
| | | 861 | | throw new InvalidOperationException( |
| | | 862 | | $"{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. |
| | | 865 | | if (value.Length > IdentifierCap) |
| | | 866 | | throw new InvalidOperationException( |
| | | 867 | | $"{nameof(SqlServerAsyncResponseChannelOptions)}.{name} '{value}' is {value.Length} characters; SQL Serv |
| | | 868 | | } |
| | | 869 | | |
| | | 870 | | private static bool IsIdentifier(string value) |
| | | 871 | | { |
| | | 872 | | if (value.Length == 0 || !(char.IsAsciiLetter(value[0]) || value[0] == '_')) |
| | | 873 | | return false; |
| | | 874 | | |
| | | 875 | | foreach (var c in value) |
| | | 876 | | { |
| | | 877 | | if (!(char.IsAsciiLetterOrDigit(c) || c == '_')) |
| | | 878 | | return false; |
| | | 879 | | } |
| | | 880 | | |
| | | 881 | | return true; |
| | | 882 | | } |
| | | 883 | | |
| | | 884 | | 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) |
| | | 895 | | => RelationalNamePlan.DerivedName(table, "_ack_seq", IdentifierCap); |
| | | 896 | | |
| | | 897 | | private static string IndexName(string table, string suffix) |
| | | 898 | | => 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 | | { |
| | | 910 | | (string Role, string Name)[] plan = |
| | | 911 | | [ |
| | | 912 | | ($"{nameof(options.RecoveryStateTable)} table", options.RecoveryStateTable), |
| | | 913 | | ($"{nameof(options.MessageTable)} table", options.MessageTable), |
| | | 914 | | ($"{nameof(options.SubscriberTable)} table", options.SubscriberTable), |
| | | 915 | | ("ack sequence (derived from MessageTable)", SequenceName(options.MessageTable)), |
| | | 916 | | ]; |
| | | 917 | | RelationalNamePlan.RequireDistinct( |
| | | 918 | | plan, |
| | | 919 | | nameof(SqlServerAsyncResponseChannelOptions), |
| | | 920 | | ". Tables and the sequence derived from MessageTable share one schema-object namespace and must be distinct |
| | | 921 | | "(long names reserve suffix space by truncating the table stem). Shorten or de-overlap the configured table |
| | | 922 | | } |
| | | 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) |
| | | 931 | | => $"DATEADD(SECOND, CAST({parameterName} / 1000 AS int), DATEADD(MILLISECOND, CAST({parameterName} % 1000 AS in |
| | | 932 | | |
| | | 933 | | 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) |
| | | 941 | | => $"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 | | { |
| | | 950 | | var interval = _options.PruneInterval; |
| | | 951 | | if (interval <= TimeSpan.Zero) |
| | | 952 | | return true; |
| | | 953 | | |
| | | 954 | | var now = DateTime.UtcNow.Ticks; |
| | | 955 | | var last = Interlocked.Read(ref lastTicks); |
| | | 956 | | return now - last >= interval.Ticks |
| | | 957 | | && Interlocked.CompareExchange(ref lastTicks, now, last) == last; |
| | | 958 | | } |
| | | 959 | | } |