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

Information
Class: AsyncResponse.Transports.PostgreSQL.PostgreSqlTransportStore
Assembly: AsyncResponse.Transports.PostgreSQL
File(s): /_/src/Transports/AsyncResponse.Transports.PostgreSQL/PostgreSqlTransportStore.cs
Line coverage
94%
Covered lines: 307
Uncovered lines: 19
Coverable lines: 326
Total lines: 618
Line coverage: 94.1%
Branch coverage
88%
Covered branches: 62
Total branches: 70
Branch coverage: 88.5%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%11100%
get_Schema()100%11100%
get_MessageTable()100%11100%
EnsureCreatedAsync()88.88%181898.8%
RelationExistsAsync()100%11100%
PublishAsync()0%2257.14%
TryClaimAsync()100%44100%
ClaimBatchAsync()100%44100%
InsertAsync()100%1212100%
AckAsync()100%11100%
RenewLeaseAsync()100%210%
NakAsync()100%11100%
DeadLetterAsync()75%181680.39%
ExecuteListenAsync()50%2290%
IsWakeFor(...)50%22100%
PruneDeadLettersIfDueAsync()100%44100%
ShouldPruneDeadLetters()100%22100%
DeserializeHeaders(...)100%11100%
Sanitize(...)100%11100%
SchemaAdvisoryLockKey(...)100%22100%
Quote(...)100%11100%
IndexName(...)100%11100%
.cctor()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>
 24internal sealed record PostgreSqlTransportDelivery(
 25    Guid Id,
 26    string Queue,
 27    string Payload,
 28    IReadOnlyDictionary<string, string> Headers,
 29    int Attempt,
 30    Func<ValueTask> AckAsync,
 31    Func<TimeSpan, ValueTask> NakAsync,
 32    Func<Exception, bool, CancellationToken, ValueTask<bool>> DeadLetterAsync,
 33    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;
 22741    private readonly SemaphoreSlim _ensureGate = new(1, 1);
 42    private bool _created;
 43    private readonly long _schemaLockKey;
 44    private long _lastDeadLetterPruneTicks;
 45
 22746    public PostgreSqlTransportStore(
 22747        NpgsqlDataSource dataSource,
 22748        IOptions<PostgreSqlAsyncResponseTransportOptions> options,
 22749        ILogger<PostgreSqlTransportStore>? logger = null)
 50    {
 22751        _dataSource = dataSource;
 22752        _options = options.Value;
 22753        _logger = logger;
 22754        PostgreSqlTransportOptionsValidator.ValidateCommon(_options);
 22755        Schema = Quote(_options.SchemaName);
 22756        MessageTable = $"{Schema}.{Quote(_options.MessageTable)}";
 22757        _schemaLockKey = SchemaAdvisoryLockKey(_options.SchemaName);
 22758    }
 59
 43760    public string Schema { get; }
 564461    public string MessageTable { get; }
 62
 63    public async Task EnsureCreatedAsync(CancellationToken cancellationToken = default)
 64    {
 331265        if (_created)
 270366            return;
 67
 60968        await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 69        try
 70        {
 60971            if (_created)
 38472                return;
 73
 22574            await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 21375            await using var transaction = await connection.BeginTransactionAsync(cancellationToken).ConfigureAwait(false
 76
 21377            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.
 20984                await using (var lockCommand = connection.CreateCommand())
 85                {
 20986                    lockCommand.Transaction = transaction;
 20987                    lockCommand.CommandText = "SELECT pg_advisory_xact_lock(@lock_key);";
 20988                    lockCommand.Parameters.AddWithValue("lock_key", _schemaLockKey);
 20989                    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.
 209102                await using var command = connection.CreateCommand();
 209103                command.Transaction = transaction;
 209104                command.CommandText =
 209105                    $"""
 209106                    CREATE SCHEMA IF NOT EXISTS {Schema};
 209107
 209108                    CREATE TABLE IF NOT EXISTS {MessageTable} (
 209109                        id uuid PRIMARY KEY,
 209110                        queue text NOT NULL,
 209111                        payload_json jsonb NOT NULL,
 209112                        headers_json jsonb NOT NULL DEFAULT jsonb_build_object(),
 209113                        created_at timestamptz NOT NULL DEFAULT now(),
 209114                        available_at timestamptz NOT NULL DEFAULT now(),
 209115                        locked_until timestamptz NULL,
 209116                        lock_id uuid NULL,
 209117                        attempts integer NOT NULL DEFAULT 0,
 209118                        dead_letter_reason text NULL
 209119                    );
 209120                    CREATE INDEX IF NOT EXISTS {Quote(IndexName(_options.MessageTable, "ready"))}
 209121                        ON {MessageTable} (queue, available_at, created_at);
 209122                    CREATE INDEX IF NOT EXISTS {Quote(IndexName(_options.MessageTable, "created"))}
 209123                        ON {MessageTable} (created_at);
 209124                    """;
 125                try
 126                {
 209127                    await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 208128                }
 1129                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".
 1135                    throw new InvalidOperationException(PostgreSqlRelationVerifier.DdlCollisionMessage("transport", _opt
 136                }
 208137            }
 4138            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.
 211160            var readyIndex = IndexName(_options.MessageTable, "ready");
 211161            var verifyReadyIndex = _options.AutoCreateSchema
 211162                || await RelationExistsAsync(connection, transaction, readyIndex, cancellationToken).ConfigureAwait(fals
 211163            if (!verifyReadyIndex)
 164            {
 2165                _logger?.LogWarning(
 2166                    "PostgreSQL transport table {Schema}.{Table} has no dequeue index {Index} and AutoCreateSchema is di
 2167                    "Claims still work, but their cost grows with the backlog — performance only; create the index over 
 2168                    "(queue, available_at, created_at) as described in docs/postgresql.md.",
 2169                    _options.SchemaName,
 2170                    _options.MessageTable,
 2171                    readyIndex);
 172            }
 173
 211174            await PostgreSqlRelationVerifier.VerifyAsync(
 211175                connection,
 211176                transaction,
 211177                _options.SchemaName,
 211178                "transport",
 211179                [
 211180                    new(_options.MessageTable, 'r', Columns:
 211181                        [
 211182                            new("id", "uuid", Nullable: false),
 211183                            new("queue", "text", Nullable: false, RequiresDeterministicCollation: true),
 211184                            new("payload_json", "jsonb", Nullable: false),
 211185                            new("headers_json", "jsonb", Nullable: false, DefaultExpression: "jsonb_build_object()"),
 211186                            new("created_at", "timestamp with time zone", Nullable: false, DefaultExpression: "now()"),
 211187                            new("available_at", "timestamp with time zone", Nullable: false, DefaultExpression: "now()")
 211188                            new("locked_until", "timestamp with time zone", Nullable: true),
 211189                            new("lock_id", "uuid", Nullable: true),
 211190                            new("attempts", "integer", Nullable: false, DefaultExpression: "0"),
 211191                            new("dead_letter_reason", "text", Nullable: true),
 211192                        ], PrimaryKey: ["id"]),
 211193                    .. verifyReadyIndex
 211194                        ? (PostgreSqlRelationVerifier.ExpectedRelation[])
 211195                            [new(readyIndex, 'i', _options.MessageTable, ["queue", "available_at", "created_at"])]
 211196                        : [],
 211197                    new(IndexName(_options.MessageTable, "created"), 'i', _options.MessageTable, ["created_at"]),
 211198                ],
 211199                cancellationToken).ConfigureAwait(false);
 200
 206201            await transaction.CommitAsync(cancellationToken).ConfigureAwait(false);
 206202            _created = true;
 206203        }
 204        finally
 205        {
 609206            _ensureGate.Release();
 207        }
 3294208    }
 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    {
 7217        await using var command = connection.CreateCommand();
 7218        command.Transaction = transaction;
 7219        command.CommandText =
 7220            """
 7221            SELECT EXISTS (
 7222                SELECT 1
 7223                FROM pg_catalog.pg_class c
 7224                JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
 7225                WHERE n.nspname = @schema AND c.relname = @table);
 7226            """;
 7227        command.Parameters.AddWithValue("schema", _options.SchemaName);
 7228        command.Parameters.AddWithValue("table", relation);
 7229        return (bool)(await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false))!;
 7230    }
 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    {
 422244        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        {
 420253            await PruneDeadLettersIfDueAsync(cancellationToken).ConfigureAwait(false);
 420254        }
 0255        catch (Exception ex)
 256        {
 0257            _logger?.LogWarning(ex, "PostgreSQL dead-letter prune failed; the publish it followed is committed and unaff
 0258        }
 420259    }
 260
 261    public async Task<PostgreSqlTransportDelivery?> TryClaimAsync(string queue, TimeSpan lockTimeout, CancellationToken 
 262    {
 2081263        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 2081264        var lockId = Guid.NewGuid();
 265
 2081266        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 2081267        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.
 2081273        command.CommandText =
 2081274            $"""
 2081275            WITH next AS (
 2081276                SELECT id
 2081277                FROM {MessageTable}
 2081278                WHERE queue = @queue
 2081279                  AND available_at <= now()
 2081280                  AND (locked_until IS NULL OR locked_until <= now())
 2081281                ORDER BY available_at, created_at
 2081282                FOR UPDATE SKIP LOCKED
 2081283                LIMIT 1
 2081284            )
 2081285            UPDATE {MessageTable} AS message
 2081286            SET attempts = message.attempts + 1,
 2081287                locked_until = now() + @lock_timeout,
 2081288                lock_id = @lock_id
 2081289            FROM next
 2081290            WHERE message.id = next.id
 2081291            RETURNING message.id, message.queue, message.payload_json::text, message.headers_json::text, message.attempt
 2081292            """;
 2081293        command.Parameters.AddWithValue("queue", queue);
 2081294        command.Parameters.AddWithValue("lock_timeout", lockTimeout);
 2081295        command.Parameters.AddWithValue("lock_id", lockId);
 296
 2081297        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 2057298        if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 1634299            return null;
 300
 423301        var id = reader.GetGuid(0);
 423302        var payload = reader.GetString(2);
 423303        var headerJson = reader.GetString(3);
 423304        var attempt = reader.GetInt32(4);
 423305        var headers = DeserializeHeaders(headerJson);
 306
 423307        return new PostgreSqlTransportDelivery(
 423308            id,
 423309            reader.GetString(1),
 423310            payload,
 423311            headers,
 423312            attempt,
 415313            () => AckAsync(id, lockId),
 4314            delay => NakAsync(id, lockId, delay),
 4315            (exception, deleteOriginal, token) => DeadLetterAsync(id, lockId, queue, payload, headers, exception, delete
 423316            () => RenewLeaseAsync(id, lockId, lockTimeout));
 2057317    }
 318
 319    public async IAsyncEnumerable<PostgreSqlTransportDelivery> ClaimBatchAsync(
 320        string queue,
 321        int batchSize,
 322        TimeSpan lockTimeout,
 323        [EnumeratorCancellation] CancellationToken cancellationToken)
 324    {
 4138325        for (var i = 0; i < batchSize; i++)
 326        {
 2067327            var delivery = await TryClaimAsync(queue, lockTimeout, cancellationToken).ConfigureAwait(false);
 2043328            if (delivery is null)
 1629329                yield break;
 414330            yield return delivery;
 331        }
 1631332    }
 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    {
 424344        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 420345        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 420346        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).
 420351        command.CommandText =
 420352            delay is null
 420353                ? $"""
 420354                  INSERT INTO {MessageTable} (id, queue, payload_json, headers_json, dead_letter_reason)
 420355                  VALUES (@id, @queue, @payload_json, @headers_json, @dead_letter_reason)
 420356                  ON CONFLICT (id) DO NOTHING;
 420357                  """
 420358                : $"""
 420359                  INSERT INTO {MessageTable} (id, queue, payload_json, headers_json, dead_letter_reason, available_at)
 420360                  VALUES (@id, @queue, @payload_json, @headers_json, @dead_letter_reason, now() + make_interval(secs => 
 420361                  ON CONFLICT (id) DO NOTHING;
 420362                  """;
 420363        if (notify)
 420364            command.CommandText += "SELECT pg_notify(@channel, @payload);";
 365
 420366        command.Parameters.AddWithValue("id", id);
 420367        command.Parameters.AddWithValue("queue", queue);
 420368        command.Parameters.Add("payload_json", NpgsqlDbType.Jsonb).Value = payload;
 420369        command.Parameters.Add("headers_json", NpgsqlDbType.Jsonb).Value = AsyncResponseJson.Serialize(headers ?? EmptyH
 420370        command.Parameters.AddWithValue("dead_letter_reason", (object?)deadLetterReason ?? DBNull.Value);
 420371        if (delay is { } pending)
 1372            command.Parameters.AddWithValue("delay_seconds", pending.TotalSeconds);
 420373        if (notify)
 374        {
 420375            command.Parameters.AddWithValue("channel", _options.NotificationChannel);
 420376            command.Parameters.AddWithValue("payload", queue);
 377        }
 378
 420379        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 420380    }
 381
 382    private async ValueTask AckAsync(Guid id, Guid lockId)
 383    {
 416384        await using var connection = await _dataSource.OpenConnectionAsync(CancellationToken.None).ConfigureAwait(false)
 416385        await using var command = connection.CreateCommand();
 416386        command.CommandText = $"DELETE FROM {MessageTable} WHERE id = @id AND lock_id = @lock_id;";
 416387        command.Parameters.AddWithValue("id", id);
 416388        command.Parameters.AddWithValue("lock_id", lockId);
 416389        await command.ExecuteNonQueryAsync(CancellationToken.None).ConfigureAwait(false);
 416390    }
 391
 392    private async ValueTask<bool> RenewLeaseAsync(Guid id, Guid lockId, TimeSpan lockTimeout)
 393    {
 0394        await using var connection = await _dataSource.OpenConnectionAsync(CancellationToken.None).ConfigureAwait(false)
 0395        await using var command = connection.CreateCommand();
 0396        command.CommandText =
 0397            $"""
 0398            UPDATE {MessageTable}
 0399            SET locked_until = now() + @lock_timeout
 0400            WHERE id = @id AND lock_id = @lock_id;
 0401            """;
 0402        command.Parameters.AddWithValue("id", id);
 0403        command.Parameters.AddWithValue("lock_id", lockId);
 0404        command.Parameters.AddWithValue("lock_timeout", lockTimeout);
 0405        return await command.ExecuteNonQueryAsync(CancellationToken.None).ConfigureAwait(false) > 0;
 0406    }
 407
 408    private async ValueTask NakAsync(Guid id, Guid lockId, TimeSpan delay)
 409    {
 4410        await using var connection = await _dataSource.OpenConnectionAsync(CancellationToken.None).ConfigureAwait(false)
 4411        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.
 4416        command.CommandText =
 4417            $"""
 4418            UPDATE {MessageTable}
 4419            SET available_at = now() + @delay,
 4420                locked_until = NULL,
 4421                lock_id = NULL
 4422            WHERE id = @id AND lock_id = @lock_id;
 4423            """;
 4424        command.Parameters.AddWithValue("id", id);
 4425        command.Parameters.AddWithValue("lock_id", lockId);
 4426        command.Parameters.AddWithValue("delay", delay);
 4427        await command.ExecuteNonQueryAsync(CancellationToken.None).ConfigureAwait(false);
 4428    }
 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    {
 8440        if (!_options.DeadLetterEnabled)
 441        {
 1442            if (deleteOriginal)
 1443                await AckAsync(id, lockId).ConfigureAwait(false);
 1444            return true;
 445        }
 446
 7447        var deadHeaders = new Dictionary<string, string>(headers, StringComparer.OrdinalIgnoreCase)
 7448        {
 7449            ["AR-DeadLetter-Reason"] = Sanitize(exception.Message),
 7450            ["AR-DeadLetter-Source-Queue"] = sourceQueue
 7451        };
 452
 453        try
 454        {
 7455            if (!deleteOriginal)
 456            {
 2457                await InsertAsync(Guid.NewGuid(), _options.DeadLetterQueue, payload, deadHeaders, exception.Message, not
 0458                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.
 5464            await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 3465            await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 3466            await using var transaction = await connection.BeginTransactionAsync(cancellationToken).ConfigureAwait(false
 3467            await using var command = connection.CreateCommand();
 3468            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.
 3474            command.CommandText =
 3475                $"""
 3476                WITH removed AS (
 3477                    DELETE FROM {MessageTable} WHERE id = @source_id AND lock_id = @lock_id RETURNING id
 3478                )
 3479                INSERT INTO {MessageTable} (id, queue, payload_json, headers_json, dead_letter_reason)
 3480                SELECT @id, @queue, @payload_json, @headers_json, @dead_letter_reason FROM removed
 3481                ON CONFLICT (id) DO NOTHING;
 3482                """;
 3483            command.Parameters.AddWithValue("id", Guid.NewGuid());
 3484            command.Parameters.AddWithValue("queue", _options.DeadLetterQueue);
 3485            command.Parameters.Add("payload_json", NpgsqlDbType.Jsonb).Value = payload;
 3486            command.Parameters.Add("headers_json", NpgsqlDbType.Jsonb).Value = AsyncResponseJson.Serialize(deadHeaders);
 3487            command.Parameters.AddWithValue("dead_letter_reason", exception.Message);
 3488            command.Parameters.AddWithValue("source_id", id);
 3489            command.Parameters.AddWithValue("lock_id", lockId);
 3490            var inserted = await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 3491            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.
 3496            if (inserted == 0)
 497            {
 1498                _logger?.LogWarning(
 1499                    "PostgreSQL dead-letter for message {MessageId} from queue {SourceQueue} no-opped: the claim's lease
 1500                    id,
 1501                    sourceQueue);
 1502                return false;
 503            }
 504
 2505            return true;
 0506        }
 4507        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.
 4511            _logger?.LogError(
 4512                ex,
 4513                "Failed to write PostgreSQL dead-letter row for message {MessageId} from queue {SourceQueue}.",
 4514                id,
 4515                sourceQueue);
 4516            return false;
 517        }
 8518    }
 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    {
 389529        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 389530        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 387531        connection.Notification += (_, e) =>
 387532        {
 606533            if (IsWakeFor(queue, e.Payload))
 606534                _ = onNotification();
 993535        };
 387536        await using (var command = connection.CreateCommand())
 537        {
 387538            command.CommandText = $"LISTEN {Quote(_options.NotificationChannel)};";
 387539            await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 540        }
 541
 992542        while (!cancellationToken.IsCancellationRequested)
 992543            await connection.WaitAsync(cancellationToken).ConfigureAwait(false);
 0544    }
 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)
 606552        => 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    {
 420561        if (_options.DeadLetterRetention is not { } retention || !ShouldPruneDeadLetters())
 416562            return;
 563
 4564        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 4565        await using var command = connection.CreateCommand();
 4566        command.CommandText = $"DELETE FROM {MessageTable} WHERE queue = @queue AND created_at < now() - @retention;";
 4567        command.Parameters.AddWithValue("queue", _options.DeadLetterQueue);
 4568        command.Parameters.AddWithValue("retention", retention);
 4569        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 420570    }
 571
 572    private bool ShouldPruneDeadLetters()
 573    {
 13574        var now = DateTime.UtcNow.Ticks;
 13575        var last = Interlocked.Read(ref _lastDeadLetterPruneTicks);
 13576        return now - last >= DeadLetterPruneThrottle.Ticks
 13577            && 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)
 435584        => DbTransportHeaders.Materialize(json);
 585
 7586    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;
 234597        var hash = offset;
 16912598        foreach (var b in Encoding.UTF8.GetBytes($"asyncresponse:ddl:{schemaName}"))
 599        {
 8222600            hash ^= b;
 8222601            hash *= prime;
 602        }
 603
 234604        return unchecked((long)hash);
 605    }
 606
 1259607    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)
 2566612        => RelationalNamePlan.DerivedName(table, $"_{suffix}_idx", identifierCap: 63);
 613
 6614    private static readonly TimeSpan DeadLetterPruneThrottle = TimeSpan.FromMinutes(1);
 615
 6616    private static readonly IReadOnlyDictionary<string, string> EmptyHeaders =
 6617        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(Npgsql.NpgsqlDataSource,Microsoft.Extensions.Options.IOptions`1<AsyncResponse.Transports.PostgreSQL.PostgreSqlAsyncResponseTransportOptions>,Microsoft.Extensions.Logging.ILogger`1<AsyncResponse.Transports.PostgreSQL.PostgreSqlTransportStore>)
get_Schema()
get_MessageTable()
EnsureCreatedAsync()
RelationExistsAsync()
PublishAsync()
TryClaimAsync()
ClaimBatchAsync()
InsertAsync()
AckAsync()
RenewLeaseAsync()
NakAsync()
DeadLetterAsync()
ExecuteListenAsync()
IsWakeFor(System.String,System.String)
PruneDeadLettersIfDueAsync()
ShouldPruneDeadLetters()
DeserializeHeaders(System.String)
Sanitize(System.String)
SchemaAdvisoryLockKey(System.String)
Quote(System.String)
IndexName(System.String,System.String)
.cctor()