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

Information
Class: AsyncResponse.Transports.PostgreSQL.PostgreSqlTransportDelivery
Assembly: AsyncResponse.Transports.PostgreSQL
File(s): /_/src/Transports/AsyncResponse.Transports.PostgreSQL/PostgreSqlTransportStore.cs
Line coverage
100%
Covered lines: 10
Uncovered lines: 0
Coverable lines: 10
Total lines: 618
Line coverage: 100%
Branch coverage
N/A
Covered branches: 0
Total branches: 0
Branch coverage: N/A
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%11100%
get_Id()100%11100%
get_Queue()100%11100%
get_Payload()100%11100%
get_Headers()100%11100%
get_Attempt()100%11100%
get_AckAsync()100%11100%
get_NakAsync()100%11100%
get_DeadLetterAsync()100%11100%
get_RenewAsync()100%11100%

File(s)

/_/src/Transports/AsyncResponse.Transports.PostgreSQL/PostgreSqlTransportStore.cs

#LineLine coverage
 1using Microsoft.Extensions.Logging;
 2using Microsoft.Extensions.Options;
 3using Npgsql;
 4using NpgsqlTypes;
 5using System.Runtime.CompilerServices;
 6using System.Text;
 7
 8using AsyncResponse.Internal;
 9
 10namespace AsyncResponse.Transports.PostgreSQL;
 11
 12internal enum PostgreSqlSubscriberRole
 13{
 14    Worker,
 15    ResponseIngress
 16}
 17
 18/// <summary>A claimed PostgreSQL transport row, decoupled from Npgsql types for dispatch tests.</summary>
 19/// <remarks>
 20/// <c>RenewAsync</c> extends the claim's lease (<c>locked_until</c>) by the original lock timeout,
 21/// fenced on the claim's <c>lock_id</c>; it returns <c>false</c> when the fence no longer matches
 22/// (the lease lapsed and another subscriber re-claimed the row).
 23/// </remarks>
 53924internal sealed record PostgreSqlTransportDelivery(
 5925    Guid Id,
 13526    string Queue,
 41527    string Payload,
 53128    IReadOnlyDictionary<string, string> Headers,
 62029    int Attempt,
 49930    Func<ValueTask> AckAsync,
 2631    Func<TimeSpan, ValueTask> NakAsync,
 4832    Func<Exception, bool, CancellationToken, ValueTask<bool>> DeadLetterAsync,
 57733    Func<ValueTask<bool>> RenewAsync);
 34
 35/// <summary>Small SQL adapter for the PostgreSQL transport queue table.</summary>
 36internal sealed class PostgreSqlTransportStore
 37{
 38    private readonly NpgsqlDataSource _dataSource;
 39    private readonly PostgreSqlAsyncResponseTransportOptions _options;
 40    private readonly ILogger<PostgreSqlTransportStore>? _logger;
 41    private readonly SemaphoreSlim _ensureGate = new(1, 1);
 42    private bool _created;
 43    private readonly long _schemaLockKey;
 44    private long _lastDeadLetterPruneTicks;
 45
 46    public PostgreSqlTransportStore(
 47        NpgsqlDataSource dataSource,
 48        IOptions<PostgreSqlAsyncResponseTransportOptions> options,
 49        ILogger<PostgreSqlTransportStore>? logger = null)
 50    {
 51        _dataSource = dataSource;
 52        _options = options.Value;
 53        _logger = logger;
 54        PostgreSqlTransportOptionsValidator.ValidateCommon(_options);
 55        Schema = Quote(_options.SchemaName);
 56        MessageTable = $"{Schema}.{Quote(_options.MessageTable)}";
 57        _schemaLockKey = SchemaAdvisoryLockKey(_options.SchemaName);
 58    }
 59
 60    public string Schema { get; }
 61    public string MessageTable { get; }
 62
 63    public async Task EnsureCreatedAsync(CancellationToken cancellationToken = default)
 64    {
 65        if (_created)
 66            return;
 67
 68        await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 69        try
 70        {
 71            if (_created)
 72                return;
 73
 74            await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 75            await using var transaction = await connection.BeginTransactionAsync(cancellationToken).ConfigureAwait(false
 76
 77            if (_options.AutoCreateSchema)
 78            {
 79                // Serialize schema creation across processes. CREATE ... IF NOT EXISTS is not atomic against a
 80                // concurrent create of the same object: two instances starting together both pass the existence
 81                // check and collide on the system catalog ("duplicate key ... pg_type_typname_nsp_index"). A
 82                // transaction-scoped advisory lock (keyed by schema, shared with the channel store) lets one
 83                // instance build the schema while the rest wait and then find it already present.
 84                await using (var lockCommand = connection.CreateCommand())
 85                {
 86                    lockCommand.Transaction = transaction;
 87                    lockCommand.CommandText = "SELECT pg_advisory_xact_lock(@lock_key);";
 88                    lockCommand.Parameters.AddWithValue("lock_key", _schemaLockKey);
 89                    await lockCommand.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 90                }
 91
 92                // The dequeue index is (queue, available_at, created_at) and the claim orders by
 93                // exactly that tail, so a claim is one ordered index descent that stops at the
 94                // first unleased row. The previous pair — an index over (queue, available_at,
 95                // locked_until, created_at) behind ORDER BY created_at — could not serve its own
 96                // ordering past the available_at range: the planner either walked the created_at
 97                // index through every older row of the OTHER logical queues (dead letters kept for
 98                // retention, delayed jobs) or sorted the whole ready set, on every claim, so
 99                // draining a burst cost its square. A table created by an older build keeps its
 100                // "<table>_claim_idx"; nothing reads it any more and nothing here drops it (DROP
 101                // INDEX needs an ACCESS EXCLUSIVE lock on a live queue) — see docs/postgresql.md.
 102                await using var command = connection.CreateCommand();
 103                command.Transaction = transaction;
 104                command.CommandText =
 105                    $"""
 106                    CREATE SCHEMA IF NOT EXISTS {Schema};
 107
 108                    CREATE TABLE IF NOT EXISTS {MessageTable} (
 109                        id uuid PRIMARY KEY,
 110                        queue text NOT NULL,
 111                        payload_json jsonb NOT NULL,
 112                        headers_json jsonb NOT NULL DEFAULT jsonb_build_object(),
 113                        created_at timestamptz NOT NULL DEFAULT now(),
 114                        available_at timestamptz NOT NULL DEFAULT now(),
 115                        locked_until timestamptz NULL,
 116                        lock_id uuid NULL,
 117                        attempts integer NOT NULL DEFAULT 0,
 118                        dead_letter_reason text NULL
 119                    );
 120                    CREATE INDEX IF NOT EXISTS {Quote(IndexName(_options.MessageTable, "ready"))}
 121                        ON {MessageTable} (queue, available_at, created_at);
 122                    CREATE INDEX IF NOT EXISTS {Quote(IndexName(_options.MessageTable, "created"))}
 123                        ON {MessageTable} (created_at);
 124                    """;
 125                try
 126                {
 127                    await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 128                }
 129                catch (PostgresException ex) when (ex.SqlState is PostgresErrorCodes.WrongObjectType or PostgresErrorCod
 130                {
 131                    // E.g. CREATE INDEX ... ON a name that is really another component's index:
 132                    // IF NOT EXISTS skipped the table create, and the dependent statement then hits
 133                    // the wrong relation kind mid-batch — surface the namespace collision instead of
 134                    // the raw "cannot open relation".
 135                    throw new InvalidOperationException(PostgreSqlRelationVerifier.DdlCollisionMessage("transport", _opt
 136                }
 137            }
 138            else if (!await RelationExistsAsync(connection, transaction, _options.MessageTable, cancellationToken).Confi
 139            {
 140                // Operator-managed schema and the migration has not run yet: the first query
 141                // surfaces a clear PostgreSQL error (the documented "create it yourself, later"
 142                // workflow), and _created stays unlatched so a later operation re-verifies once
 143                // the migration lands. When the relation DOES exist it flows into the same catalog
 144                // verification the DDL path uses — operator-provisioned schemas are exactly what
 145                // that check exists for.
 146                return;
 147            }
 148
 149            // The transport can share a schema with the channel and durable-flow stores (and
 150            // unrelated objects), whose derived names its own validation cannot see — and
 151            // IF NOT EXISTS also accepts a same-name index with the WRONG definition, exactly as
 152            // an operator-provisioned table can carry the wrong shape. Verify against the catalog
 153            // that every relation actually IS what this store reads and writes, definitions
 154            // included (in-transaction, under the shared DDL lock when this build just ran the DDL).
 155            //
 156            // The dequeue index is REQUIRED only where this build's DDL just guaranteed it. On an
 157            // operator-managed schema it is verified when present and only warned about when
 158            // absent: it is claim performance, not correctness, and a migration written for an
 159            // older build (which carried "<table>_claim_idx" instead) must not fail startup over it.
 160            var readyIndex = IndexName(_options.MessageTable, "ready");
 161            var verifyReadyIndex = _options.AutoCreateSchema
 162                || await RelationExistsAsync(connection, transaction, readyIndex, cancellationToken).ConfigureAwait(fals
 163            if (!verifyReadyIndex)
 164            {
 165                _logger?.LogWarning(
 166                    "PostgreSQL transport table {Schema}.{Table} has no dequeue index {Index} and AutoCreateSchema is di
 167                    "Claims still work, but their cost grows with the backlog — performance only; create the index over 
 168                    "(queue, available_at, created_at) as described in docs/postgresql.md.",
 169                    _options.SchemaName,
 170                    _options.MessageTable,
 171                    readyIndex);
 172            }
 173
 174            await PostgreSqlRelationVerifier.VerifyAsync(
 175                connection,
 176                transaction,
 177                _options.SchemaName,
 178                "transport",
 179                [
 180                    new(_options.MessageTable, 'r', Columns:
 181                        [
 182                            new("id", "uuid", Nullable: false),
 183                            new("queue", "text", Nullable: false, RequiresDeterministicCollation: true),
 184                            new("payload_json", "jsonb", Nullable: false),
 185                            new("headers_json", "jsonb", Nullable: false, DefaultExpression: "jsonb_build_object()"),
 186                            new("created_at", "timestamp with time zone", Nullable: false, DefaultExpression: "now()"),
 187                            new("available_at", "timestamp with time zone", Nullable: false, DefaultExpression: "now()")
 188                            new("locked_until", "timestamp with time zone", Nullable: true),
 189                            new("lock_id", "uuid", Nullable: true),
 190                            new("attempts", "integer", Nullable: false, DefaultExpression: "0"),
 191                            new("dead_letter_reason", "text", Nullable: true),
 192                        ], PrimaryKey: ["id"]),
 193                    .. verifyReadyIndex
 194                        ? (PostgreSqlRelationVerifier.ExpectedRelation[])
 195                            [new(readyIndex, 'i', _options.MessageTable, ["queue", "available_at", "created_at"])]
 196                        : [],
 197                    new(IndexName(_options.MessageTable, "created"), 'i', _options.MessageTable, ["created_at"]),
 198                ],
 199                cancellationToken).ConfigureAwait(false);
 200
 201            await transaction.CommitAsync(cancellationToken).ConfigureAwait(false);
 202            _created = true;
 203        }
 204        finally
 205        {
 206            _ensureGate.Release();
 207        }
 208    }
 209
 210    /// <summary>
 211    /// Reports whether ANY relation occupies the given name in the configured schema (any relkind:
 212    /// a view or foreign component's object must reach verification, which names the precise
 213    /// wrong-kind reason instead of skipping the checks).
 214    /// </summary>
 215    private async Task<bool> RelationExistsAsync(NpgsqlConnection connection, NpgsqlTransaction transaction, string rela
 216    {
 217        await using var command = connection.CreateCommand();
 218        command.Transaction = transaction;
 219        command.CommandText =
 220            """
 221            SELECT EXISTS (
 222                SELECT 1
 223                FROM pg_catalog.pg_class c
 224                JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
 225                WHERE n.nspname = @schema AND c.relname = @table);
 226            """;
 227        command.Parameters.AddWithValue("schema", _options.SchemaName);
 228        command.Parameters.AddWithValue("table", relation);
 229        return (bool)(await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false))!;
 230    }
 231
 232    /// <summary>
 233    /// Publishes a queue row. The caller supplies the id so a retried publish is idempotent
 234    /// (<c>ON CONFLICT DO NOTHING</c>) rather than inserting a duplicate job.
 235    /// </summary>
 236    public async Task PublishAsync(
 237        Guid id,
 238        string queue,
 239        string payload,
 240        IReadOnlyDictionary<string, string>? headers,
 241        CancellationToken cancellationToken,
 242        TimeSpan? delay = null)
 243    {
 244        await InsertAsync(id, queue, payload, headers, deadLetterReason: null, notify: true, cancellationToken, delay).C
 245
 246        // The row is committed: nothing after this line may fail the publish. A prune that threw
 247        // (a lock timeout, a dropped connection, the caller's token firing mid-DELETE) reported a
 248        // FAILED publish for a job that is already claimable, and the caller's retry inserts it
 249        // again once a subscriber has consumed and deleted the first copy — one job, run twice.
 250        // The prune is opportunistic housekeeping; the next throttle window retries it.
 251        try
 252        {
 253            await PruneDeadLettersIfDueAsync(cancellationToken).ConfigureAwait(false);
 254        }
 255        catch (Exception ex)
 256        {
 257            _logger?.LogWarning(ex, "PostgreSQL dead-letter prune failed; the publish it followed is committed and unaff
 258        }
 259    }
 260
 261    public async Task<PostgreSqlTransportDelivery?> TryClaimAsync(string queue, TimeSpan lockTimeout, CancellationToken 
 262    {
 263        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 264        var lockId = Guid.NewGuid();
 265
 266        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 267        await using var command = connection.CreateCommand();
 268        // Ready order is AVAILABILITY order — (available_at, created_at), the dequeue index's own
 269        // key order behind the queue equality — so the scan returns its first unleased row without
 270        // sorting. It equals publish order for every row that was neither delayed nor NAKed (both
 271        // columns default to the same now()); a delayed or redelivered row queues by when it became
 272        // due instead of jumping ahead of everything published while it waited.
 273        command.CommandText =
 274            $"""
 275            WITH next AS (
 276                SELECT id
 277                FROM {MessageTable}
 278                WHERE queue = @queue
 279                  AND available_at <= now()
 280                  AND (locked_until IS NULL OR locked_until <= now())
 281                ORDER BY available_at, created_at
 282                FOR UPDATE SKIP LOCKED
 283                LIMIT 1
 284            )
 285            UPDATE {MessageTable} AS message
 286            SET attempts = message.attempts + 1,
 287                locked_until = now() + @lock_timeout,
 288                lock_id = @lock_id
 289            FROM next
 290            WHERE message.id = next.id
 291            RETURNING message.id, message.queue, message.payload_json::text, message.headers_json::text, message.attempt
 292            """;
 293        command.Parameters.AddWithValue("queue", queue);
 294        command.Parameters.AddWithValue("lock_timeout", lockTimeout);
 295        command.Parameters.AddWithValue("lock_id", lockId);
 296
 297        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 298        if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 299            return null;
 300
 301        var id = reader.GetGuid(0);
 302        var payload = reader.GetString(2);
 303        var headerJson = reader.GetString(3);
 304        var attempt = reader.GetInt32(4);
 305        var headers = DeserializeHeaders(headerJson);
 306
 307        return new PostgreSqlTransportDelivery(
 308            id,
 309            reader.GetString(1),
 310            payload,
 311            headers,
 312            attempt,
 313            () => AckAsync(id, lockId),
 314            delay => NakAsync(id, lockId, delay),
 315            (exception, deleteOriginal, token) => DeadLetterAsync(id, lockId, queue, payload, headers, exception, delete
 316            () => RenewLeaseAsync(id, lockId, lockTimeout));
 317    }
 318
 319    public async IAsyncEnumerable<PostgreSqlTransportDelivery> ClaimBatchAsync(
 320        string queue,
 321        int batchSize,
 322        TimeSpan lockTimeout,
 323        [EnumeratorCancellation] CancellationToken cancellationToken)
 324    {
 325        for (var i = 0; i < batchSize; i++)
 326        {
 327            var delivery = await TryClaimAsync(queue, lockTimeout, cancellationToken).ConfigureAwait(false);
 328            if (delivery is null)
 329                yield break;
 330            yield return delivery;
 331        }
 332    }
 333
 334    private async Task InsertAsync(
 335        Guid id,
 336        string queue,
 337        string payload,
 338        IReadOnlyDictionary<string, string>? headers,
 339        string? deadLetterReason,
 340        bool notify,
 341        CancellationToken cancellationToken,
 342        TimeSpan? delay = null)
 343    {
 344        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 345        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 346        await using var command = connection.CreateCommand();
 347        // Native delayed delivery: available_at gates the claim query, and the delay arithmetic
 348        // runs on the DATABASE clock (now() + interval), matching the claim-side now() so client
 349        // clock skew cannot shift the due time. A due row is picked up by the subscriber's next
 350        // poll tick (EmptyPollDelay bounds the extra latency).
 351        command.CommandText =
 352            delay is null
 353                ? $"""
 354                  INSERT INTO {MessageTable} (id, queue, payload_json, headers_json, dead_letter_reason)
 355                  VALUES (@id, @queue, @payload_json, @headers_json, @dead_letter_reason)
 356                  ON CONFLICT (id) DO NOTHING;
 357                  """
 358                : $"""
 359                  INSERT INTO {MessageTable} (id, queue, payload_json, headers_json, dead_letter_reason, available_at)
 360                  VALUES (@id, @queue, @payload_json, @headers_json, @dead_letter_reason, now() + make_interval(secs => 
 361                  ON CONFLICT (id) DO NOTHING;
 362                  """;
 363        if (notify)
 364            command.CommandText += "SELECT pg_notify(@channel, @payload);";
 365
 366        command.Parameters.AddWithValue("id", id);
 367        command.Parameters.AddWithValue("queue", queue);
 368        command.Parameters.Add("payload_json", NpgsqlDbType.Jsonb).Value = payload;
 369        command.Parameters.Add("headers_json", NpgsqlDbType.Jsonb).Value = AsyncResponseJson.Serialize(headers ?? EmptyH
 370        command.Parameters.AddWithValue("dead_letter_reason", (object?)deadLetterReason ?? DBNull.Value);
 371        if (delay is { } pending)
 372            command.Parameters.AddWithValue("delay_seconds", pending.TotalSeconds);
 373        if (notify)
 374        {
 375            command.Parameters.AddWithValue("channel", _options.NotificationChannel);
 376            command.Parameters.AddWithValue("payload", queue);
 377        }
 378
 379        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 380    }
 381
 382    private async ValueTask AckAsync(Guid id, Guid lockId)
 383    {
 384        await using var connection = await _dataSource.OpenConnectionAsync(CancellationToken.None).ConfigureAwait(false)
 385        await using var command = connection.CreateCommand();
 386        command.CommandText = $"DELETE FROM {MessageTable} WHERE id = @id AND lock_id = @lock_id;";
 387        command.Parameters.AddWithValue("id", id);
 388        command.Parameters.AddWithValue("lock_id", lockId);
 389        await command.ExecuteNonQueryAsync(CancellationToken.None).ConfigureAwait(false);
 390    }
 391
 392    private async ValueTask<bool> RenewLeaseAsync(Guid id, Guid lockId, TimeSpan lockTimeout)
 393    {
 394        await using var connection = await _dataSource.OpenConnectionAsync(CancellationToken.None).ConfigureAwait(false)
 395        await using var command = connection.CreateCommand();
 396        command.CommandText =
 397            $"""
 398            UPDATE {MessageTable}
 399            SET locked_until = now() + @lock_timeout
 400            WHERE id = @id AND lock_id = @lock_id;
 401            """;
 402        command.Parameters.AddWithValue("id", id);
 403        command.Parameters.AddWithValue("lock_id", lockId);
 404        command.Parameters.AddWithValue("lock_timeout", lockTimeout);
 405        return await command.ExecuteNonQueryAsync(CancellationToken.None).ConfigureAwait(false) > 0;
 406    }
 407
 408    private async ValueTask NakAsync(Guid id, Guid lockId, TimeSpan delay)
 409    {
 410        await using var connection = await _dataSource.OpenConnectionAsync(CancellationToken.None).ConfigureAwait(false)
 411        await using var command = connection.CreateCommand();
 412        // No NOTIFY: the released row only becomes claimable once its delay has passed
 413        // (RedeliveryDelay is validated positive), so a wake sent now made every idle subscriber
 414        // of every queue, in every process, poll for a row none of them could claim yet — on each
 415        // handler failure. The row is picked up by the first poll tick after it falls due.
 416        command.CommandText =
 417            $"""
 418            UPDATE {MessageTable}
 419            SET available_at = now() + @delay,
 420                locked_until = NULL,
 421                lock_id = NULL
 422            WHERE id = @id AND lock_id = @lock_id;
 423            """;
 424        command.Parameters.AddWithValue("id", id);
 425        command.Parameters.AddWithValue("lock_id", lockId);
 426        command.Parameters.AddWithValue("delay", delay);
 427        await command.ExecuteNonQueryAsync(CancellationToken.None).ConfigureAwait(false);
 428    }
 429
 430    private async ValueTask<bool> DeadLetterAsync(
 431        Guid id,
 432        Guid lockId,
 433        string sourceQueue,
 434        string payload,
 435        IReadOnlyDictionary<string, string> headers,
 436        Exception exception,
 437        bool deleteOriginal,
 438        CancellationToken cancellationToken)
 439    {
 440        if (!_options.DeadLetterEnabled)
 441        {
 442            if (deleteOriginal)
 443                await AckAsync(id, lockId).ConfigureAwait(false);
 444            return true;
 445        }
 446
 447        var deadHeaders = new Dictionary<string, string>(headers, StringComparer.OrdinalIgnoreCase)
 448        {
 449            ["AR-DeadLetter-Reason"] = Sanitize(exception.Message),
 450            ["AR-DeadLetter-Source-Queue"] = sourceQueue
 451        };
 452
 453        try
 454        {
 455            if (!deleteOriginal)
 456            {
 457                await InsertAsync(Guid.NewGuid(), _options.DeadLetterQueue, payload, deadHeaders, exception.Message, not
 458                return true;
 459            }
 460
 461            // The DLQ insert and the original-row delete must commit atomically: split across two
 462            // connections, a crash between them leaves the original row to be redelivered and
 463            // dead-lettered again, duplicating the DLQ entry.
 464            await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 465            await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 466            await using var transaction = await connection.BeginTransactionAsync(cancellationToken).ConfigureAwait(false
 467            await using var command = connection.CreateCommand();
 468            command.Transaction = transaction;
 469            // The DLQ row is written ONLY if the fenced delete matched. A stale claim (the lease
 470            // lapsed and a peer re-claimed the row) must no-op here exactly as the fenced ack and
 471            // NAK do; writing the row unconditionally buried a full copy of a message that is still
 472            // live and may yet succeed under its new owner, so the DLQ showed a poison entry for
 473            // work that completed — and an operator replaying it duplicated its side effects.
 474            command.CommandText =
 475                $"""
 476                WITH removed AS (
 477                    DELETE FROM {MessageTable} WHERE id = @source_id AND lock_id = @lock_id RETURNING id
 478                )
 479                INSERT INTO {MessageTable} (id, queue, payload_json, headers_json, dead_letter_reason)
 480                SELECT @id, @queue, @payload_json, @headers_json, @dead_letter_reason FROM removed
 481                ON CONFLICT (id) DO NOTHING;
 482                """;
 483            command.Parameters.AddWithValue("id", Guid.NewGuid());
 484            command.Parameters.AddWithValue("queue", _options.DeadLetterQueue);
 485            command.Parameters.Add("payload_json", NpgsqlDbType.Jsonb).Value = payload;
 486            command.Parameters.Add("headers_json", NpgsqlDbType.Jsonb).Value = AsyncResponseJson.Serialize(deadHeaders);
 487            command.Parameters.AddWithValue("dead_letter_reason", exception.Message);
 488            command.Parameters.AddWithValue("source_id", id);
 489            command.Parameters.AddWithValue("lock_id", lockId);
 490            var inserted = await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 491            await transaction.CommitAsync(cancellationToken).ConfigureAwait(false);
 492
 493            // Zero rows means the fence was lost, not that the write failed. Report it as a
 494            // non-dead-letter so the caller does not log a burial that did not happen; its NAK
 495            // fallback is fenced too, so the new owner keeps the row untouched.
 496            if (inserted == 0)
 497            {
 498                _logger?.LogWarning(
 499                    "PostgreSQL dead-letter for message {MessageId} from queue {SourceQueue} no-opped: the claim's lease
 500                    id,
 501                    sourceQueue);
 502                return false;
 503            }
 504
 505            return true;
 506        }
 507        catch (Exception ex)
 508        {
 509            // Callers decide the redelivery consequence from the false return; log the cause here so
 510            // a failing dead-letter write is never silent.
 511            _logger?.LogError(
 512                ex,
 513                "Failed to write PostgreSQL dead-letter row for message {MessageId} from queue {SourceQueue}.",
 514                id,
 515                sourceQueue);
 516            return false;
 517        }
 518    }
 519
 520    /// <summary>
 521    /// LISTENs on the transport's notification channel and invokes <paramref name="onNotification"/>
 522    /// for each wake. Every logical queue of every process shares the one channel and a publish
 523    /// NOTIFYs the queue it inserted into, so a listener that names its <paramref name="queue"/>
 524    /// is woken only for that queue; without it (<c>null</c>) every publish to any queue wakes it
 525    /// into a claim that finds nothing.
 526    /// </summary>
 527    public async Task ExecuteListenAsync(Func<Task> onNotification, CancellationToken cancellationToken, string? queue =
 528    {
 529        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 530        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 531        connection.Notification += (_, e) =>
 532        {
 533            if (IsWakeFor(queue, e.Payload))
 534                _ = onNotification();
 535        };
 536        await using (var command = connection.CreateCommand())
 537        {
 538            command.CommandText = $"LISTEN {Quote(_options.NotificationChannel)};";
 539            await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 540        }
 541
 542        while (!cancellationToken.IsCancellationRequested)
 543            await connection.WaitAsync(cancellationToken).ConfigureAwait(false);
 544    }
 545
 546    /// <summary>
 547    /// Whether a NOTIFY payload is a wake for <paramref name="queue"/>. The payload is the queue
 548    /// name a publish inserted into, compared ordinally like the queue column itself; a listener
 549    /// that names no queue keeps every wake.
 550    /// </summary>
 551    internal static bool IsWakeFor(string? queue, string payload)
 552        => queue is null || string.Equals(payload, queue, StringComparison.Ordinal);
 553
 554    /// <summary>
 555    /// Opportunistically deletes dead-letter rows older than the configured retention. No-op unless
 556    /// <see cref="PostgreSqlAsyncResponseTransportOptions.DeadLetterRetention"/> is set, and throttled
 557    /// so the DELETE runs at most once per minute regardless of publish rate.
 558    /// </summary>
 559    private async Task PruneDeadLettersIfDueAsync(CancellationToken cancellationToken)
 560    {
 561        if (_options.DeadLetterRetention is not { } retention || !ShouldPruneDeadLetters())
 562            return;
 563
 564        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 565        await using var command = connection.CreateCommand();
 566        command.CommandText = $"DELETE FROM {MessageTable} WHERE queue = @queue AND created_at < now() - @retention;";
 567        command.Parameters.AddWithValue("queue", _options.DeadLetterQueue);
 568        command.Parameters.AddWithValue("retention", retention);
 569        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 570    }
 571
 572    private bool ShouldPruneDeadLetters()
 573    {
 574        var now = DateTime.UtcNow.Ticks;
 575        var last = Interlocked.Read(ref _lastDeadLetterPruneTicks);
 576        return now - last >= DeadLetterPruneThrottle.Ticks
 577            && Interlocked.CompareExchange(ref _lastDeadLetterPruneTicks, now, last) == last;
 578    }
 579
 580    // Lenient by contract (see DbTransportHeaders): this runs after the claim already committed
 581    // attempts+1/lock_id, so rejecting any JSON the column legally holds would create an
 582    // unkillable poison row.
 583    private static IReadOnlyDictionary<string, string> DeserializeHeaders(string json)
 584        => DbTransportHeaders.Materialize(json);
 585
 586    private static string Sanitize(string value) => value.Replace('\r', ' ').Replace('\n', ' ');
 587
 588    /// <summary>
 589    /// Stable 64-bit advisory-lock key for serializing schema creation. Must be byte-for-byte identical
 590    /// to the channel store's algorithm/discriminator so that, for a shared schema, the channel and
 591    /// transport take the same lock and never race each other on CREATE SCHEMA.
 592    /// </summary>
 593    internal static long SchemaAdvisoryLockKey(string schemaName)
 594    {
 595        const ulong offset = 14695981039346656037UL;
 596        const ulong prime = 1099511628211UL;
 597        var hash = offset;
 598        foreach (var b in Encoding.UTF8.GetBytes($"asyncresponse:ddl:{schemaName}"))
 599        {
 600            hash ^= b;
 601            hash *= prime;
 602        }
 603
 604        return unchecked((long)hash);
 605    }
 606
 607    private static string Quote(string identifier) => "\"" + identifier + "\"";
 608
 609    // Suffix space is RESERVED before capping; see RelationalNamePlan.DerivedName for why and for
 610    // the single implementation this and the SQL Server / channel stores all share.
 611    internal static string IndexName(string table, string suffix)
 612        => RelationalNamePlan.DerivedName(table, $"_{suffix}_idx", identifierCap: 63);
 613
 614    private static readonly TimeSpan DeadLetterPruneThrottle = TimeSpan.FromMinutes(1);
 615
 616    private static readonly IReadOnlyDictionary<string, string> EmptyHeaders =
 617        new Dictionary<string, string>(0, StringComparer.OrdinalIgnoreCase);
 618}
Generated by: ReportGenerator 5.5.11.0
09/22/2026 - 09:41:04
GitHub | reportgenerator.io

Methods/Properties

.ctor(System.Guid,System.String,System.String,System.Collections.Generic.IReadOnlyDictionary`2<System.String,System.String>,System.Int32,System.Func`1<System.Threading.Tasks.ValueTask>,System.Func`2<System.TimeSpan,System.Threading.Tasks.ValueTask>,System.Func`4<System.Exception,System.Boolean,System.Threading.CancellationToken,System.Threading.Tasks.ValueTask`1<System.Boolean>>,System.Func`1<System.Threading.Tasks.ValueTask`1<System.Boolean>>)
get_Id()
get_Queue()
get_Payload()
get_Headers()
get_Attempt()
get_AckAsync()
get_NakAsync()
get_DeadLetterAsync()
get_RenewAsync()