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

Information
Class: AsyncResponse.Channels.PostgreSQL.PostgreSqlChannelSql
Assembly: AsyncResponse.Channels.PostgreSQL
File(s): /_/src/Channels/AsyncResponse.Channels.PostgreSQL/PostgreSqlChannelSql.cs
Line coverage
98%
Covered lines: 552
Uncovered lines: 7
Coverable lines: 559
Total lines: 920
Line coverage: 98.7%
Branch coverage
88%
Covered branches: 113
Total branches: 128
Branch coverage: 88.2%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%11100%
get_Schema()100%11100%
get_RecoveryTable()100%11100%
get_MessageTable()100%11100%
get_SubscriberTable()100%11100%
get_AckSequence()100%11100%
get_AckSequenceName()100%11100%
get_NotificationChannel()100%11100%
EnsureCreatedAsync()83.33%6698.94%
VerifyRelationsAsync(...)100%11100%
ValidateManagedSchemaAsync()68.75%161697.14%
SaveRecoveryStateAsync()100%11100%
LoadRecoveryStatesAsync()100%66100%
DeleteRecoveryStateAsync()100%11100%
ScanRecoveryStateJsonAsync()100%1010100%
InsertMessageAsync(...)100%11100%
InsertMessageOnceAsync()92.85%141495.23%
LoadMessagesAsync()83.33%66100%
LoadMessagesByIdAsync()50%4494.44%
ReadMessagesAsync()100%88100%
TryClaimForDeliveryAsync()100%22100%
TryClaimForRecoveryAsync()100%22100%
GetSubscriptionStartAsync()100%22100%
GetServerTimeUtcAsync()50%4483.33%
IsMessageAcknowledgedAsync()50%22100%
UpsertSubscriberAsync()100%22100%
HeartbeatSubscribersAsync()100%44100%
DeleteSubscriberAsync()100%11100%
CountActiveSubscribersAsync()75%44100%
ExecuteListenAsync()50%2287.5%
PruneExpiredRecoveryAsync()100%44100%
PruneExpiredMessagesAsync()100%11100%
PruneExpiredSubscribersAsync()75%44100%
ValidateIdentifier(...)100%66100%
IsIdentifier(...)100%1212100%
Quote(...)100%11100%
SqlLiteral(...)100%11100%
IndexName(...)100%11100%
SequenceName(...)100%11100%
ValidateNamePlan(...)100%11100%
NotifyPayload(...)100%22100%
IsTransient(...)100%11100%
SchemaAdvisoryLockKey(...)100%22100%
ShouldPrune(...)100%44100%

File(s)

/_/src/Channels/AsyncResponse.Channels.PostgreSQL/PostgreSqlChannelSql.cs

#LineLine coverage
 1using Npgsql;
 2using NpgsqlTypes;
 3using System.Text;
 4
 5using AsyncResponse.Internal;
 6
 7namespace AsyncResponse.Channels.PostgreSQL;
 8
 9/// <summary>One stored response envelope row/document as the channel store returns it.</summary>
 10/// <remarks>
 11/// <c>EnvelopeJson</c> is the stored envelope, or <c>null</c> for a row the dispatch sweep loaded header-only (an
 12/// already-acknowledged row — see <see cref="PostgreSqlChannelSql.LoadMessagesAsync"/>); the
 13/// sweep hydrates the few such rows it still has to deliver through
 14/// <see cref="PostgreSqlChannelSql.LoadMessagesByIdAsync"/> before handing them to a waiter.
 15/// </remarks>
 16internal readonly record struct PostgreSqlChannelMessage(
 17    Guid Id,
 18    string CorrelationId,
 19    string? EnvelopeJson,
 20    DateTimeOffset CreatedAtUtc,
 21    DateTimeOffset? AckedAtUtc = null,
 22    long? AckedSeq = null);
 23
 24/// <summary>SQL helper for the PostgreSQL channel tables and notification channel.</summary>
 25internal sealed class PostgreSqlChannelSql
 26{
 27    // PostgreSQL rejects a NOTIFY payload of 8000 bytes or more; stay well under it. A correlation
 28    // id longer than this is sent as an empty payload, which the listener treats as "scan all".
 29    private const int MaxNotifyPayloadBytes = 7000;
 30
 31    private readonly NpgsqlDataSource _dataSource;
 32    private readonly PostgreSqlAsyncResponseChannelOptions _options;
 44433    private readonly SemaphoreSlim _ensureGate = new(1, 1);
 34    private bool _created;
 35    private readonly long _schemaLockKey;
 36    private long _lastRecoveryPruneTicks;
 37    private long _lastMessagePruneTicks;
 38    private long _lastSubscriberPruneTicks;
 39
 44440    public PostgreSqlChannelSql(NpgsqlDataSource dataSource, Microsoft.Extensions.Options.IOptions<PostgreSqlAsyncRespon
 41    {
 44442        _dataSource = dataSource;
 44443        _options = options.Value;
 44444        _options.Validate();
 45
 44446        Schema = Quote(_options.SchemaName);
 44447        RecoveryTable = $"{Schema}.{Quote(_options.RecoveryStateTable)}";
 44448        MessageTable = $"{Schema}.{Quote(_options.MessageTable)}";
 44449        SubscriberTable = $"{Schema}.{Quote(_options.SubscriberTable)}";
 44450        AckSequenceName = SequenceName(_options.MessageTable);
 44451        AckSequence = $"{Schema}.{Quote(AckSequenceName)}";
 44452        _schemaLockKey = SchemaAdvisoryLockKey(_options.SchemaName);
 44453    }
 54
 215955    public string Schema { get; }
 207456    public string RecoveryTable { get; }
 520257    public string MessageTable { get; }
 327058    public string SubscriberTable { get; }
 59
 60    /// <summary>
 61    /// Qualified name of the monotonic ack sequence. Delivery claims and subscription
 62    /// registrations draw from this ONE sequence, giving <c>acked_seq</c> and a subscription's
 63    /// start position a total order no pair of same-tick timestamps has.
 64    /// </summary>
 133265    public string AckSequence { get; }
 66
 67    /// <summary>Unquoted sequence identifier, for catalog queries.</summary>
 84268    public string AckSequenceName { get; }
 96069    public string NotificationChannel => _options.NotificationChannel;
 70
 71    public async Task EnsureCreatedAsync(CancellationToken cancellationToken = default)
 72    {
 599573        if (_created)
 555774            return;
 75
 43876        if (!_options.AutoCreateSchema)
 77        {
 78            // Manually managed schemas get a one-time validation instead of DDL: 1.0.0 added
 79            // acked_seq and its sequence, which waiter registration and delivery claims require
 80            // unconditionally — without this check an un-migrated schema fails later with a raw
 81            // "column does not exist" mid-operation instead of an actionable startup error
 82            // carrying the exact migration.
 5483            await ValidateManagedSchemaAsync(cancellationToken).ConfigureAwait(false);
 284            return;
 85        }
 86
 38487        await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 88        try
 89        {
 38490            if (_created)
 091                return;
 92
 38493            await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 38294            await using var transaction = await connection.BeginTransactionAsync(cancellationToken).ConfigureAwait(false
 95
 96            // Serialize schema creation across processes. CREATE ... IF NOT EXISTS is not atomic against a
 97            // concurrent create of the same object: two instances starting together both pass the existence
 98            // check and collide on the system catalog ("duplicate key ... pg_type_typname_nsp_index"). A
 99            // transaction-scoped advisory lock (keyed by schema, shared with the transport store) lets one
 100            // instance build the schema while the rest wait and then find it already present.
 382101            await using (var lockCommand = connection.CreateCommand())
 102            {
 382103                lockCommand.Transaction = transaction;
 382104                lockCommand.CommandText = "SELECT pg_advisory_xact_lock(@lock_key);";
 382105                lockCommand.Parameters.AddWithValue("lock_key", _schemaLockKey);
 382106                await lockCommand.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 107            }
 108
 382109            await using var command = connection.CreateCommand();
 382110            command.Transaction = transaction;
 382111            command.CommandText =
 382112                $"""
 382113                CREATE SCHEMA IF NOT EXISTS {Schema};
 382114
 382115                CREATE TABLE IF NOT EXISTS {RecoveryTable} (
 382116                    correlation_id text NOT NULL,
 382117                    registration_id uuid NOT NULL,
 382118                    state_json text NOT NULL,
 382119                    expires_at timestamptz NOT NULL,
 382120                    registered_at timestamptz NOT NULL DEFAULT now(),
 382121                    PRIMARY KEY (correlation_id, registration_id)
 382122                );
 382123                CREATE INDEX IF NOT EXISTS {Quote(IndexName(_options.RecoveryStateTable, "expires"))}
 382124                    ON {RecoveryTable} (expires_at);
 382125
 382126                CREATE TABLE IF NOT EXISTS {MessageTable} (
 382127                    id uuid PRIMARY KEY,
 382128                    correlation_id text NOT NULL,
 382129                    envelope_json text NOT NULL,
 382130                    created_at timestamptz NOT NULL DEFAULT now(),
 382131                    expires_at timestamptz NOT NULL,
 382132                    acked_at timestamptz NULL,
 382133                    acked_seq bigint NULL,
 382134                    recovery_claimed boolean NOT NULL DEFAULT false
 382135                );
 382136                ALTER TABLE {MessageTable} ADD COLUMN IF NOT EXISTS recovery_claimed boolean NOT NULL DEFAULT false;
 382137                ALTER TABLE {MessageTable} ADD COLUMN IF NOT EXISTS acked_seq bigint NULL;
 382138
 382139                -- jsonb REJECTS the \u0000 escape that System.Text.Json emits for U+0000 (SQLSTATE
 382140                -- 22P05), so any payload, exception message, propagated context value or callback
 382141                -- argument containing a NUL was unpublishable on PostgreSQL alone while every other
 382142                -- channel delivered it. Nothing here ever queries INSIDE the document — both columns
 382143                -- are read back with ::text — so text costs nothing and accepts the whole contract.
 382144                -- Guarded so the rewrite happens once, not on every start.
 382145                DO $$
 382146                BEGIN
 382147                    IF EXISTS (
 382148                        SELECT 1 FROM information_schema.columns
 382149                        WHERE table_schema = {SqlLiteral(_options.SchemaName)} AND table_name = {SqlLiteral(_options.Mes
 382150                          AND column_name = 'envelope_json' AND data_type = 'jsonb')
 382151                    THEN
 382152                        ALTER TABLE {MessageTable} ALTER COLUMN envelope_json TYPE text USING envelope_json::text;
 382153                    END IF;
 382154
 382155                    IF EXISTS (
 382156                        SELECT 1 FROM information_schema.columns
 382157                        WHERE table_schema = {SqlLiteral(_options.SchemaName)} AND table_name = {SqlLiteral(_options.Rec
 382158                          AND column_name = 'state_json' AND data_type = 'jsonb')
 382159                    THEN
 382160                        ALTER TABLE {RecoveryTable} ALTER COLUMN state_json TYPE text USING state_json::text;
 382161                    END IF;
 382162                END $$;
 382163                CREATE SEQUENCE IF NOT EXISTS {AckSequence} AS bigint;
 382164                CREATE INDEX IF NOT EXISTS {Quote(IndexName(_options.MessageTable, "correlation_created"))}
 382165                    ON {MessageTable} (correlation_id, created_at);
 382166                CREATE INDEX IF NOT EXISTS {Quote(IndexName(_options.MessageTable, "expires"))}
 382167                    ON {MessageTable} (expires_at);
 382168
 382169                CREATE TABLE IF NOT EXISTS {SubscriberTable} (
 382170                    correlation_id text NOT NULL,
 382171                    registration_id uuid NOT NULL,
 382172                    instance_id text NOT NULL,
 382173                    expires_at timestamptz NOT NULL,
 382174                    PRIMARY KEY (correlation_id, registration_id)
 382175                );
 382176                CREATE INDEX IF NOT EXISTS {Quote(IndexName(_options.SubscriberTable, "expires"))}
 382177                    ON {SubscriberTable} (expires_at);
 382178                """;
 179            try
 180            {
 382181                await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 380182            }
 2183            catch (PostgresException ex) when (ex.SqlState is PostgresErrorCodes.WrongObjectType or PostgresErrorCodes.U
 184            {
 185                // E.g. CREATE INDEX ... ON a name that is really another component's index:
 186                // IF NOT EXISTS skipped the table create, and the dependent statement then hits
 187                // the wrong relation kind mid-batch — surface the namespace collision instead of
 188                // the raw "cannot open relation".
 2189                throw new InvalidOperationException(PostgreSqlRelationVerifier.DdlCollisionMessage("channel", _options.S
 190            }
 191
 192            // Options-level ValidateNamePlan keeps THIS component's names distinct, but the
 193            // channel can share a schema with the transport and durable-flow stores (and
 194            // unrelated objects), whose derived names it cannot see — and IF NOT EXISTS also
 195            // accepts a same-name index with the WRONG definition. Verify against the catalog,
 196            // in-transaction under the shared DDL lock, that every relation actually IS what the
 197            // DDL above intended, definitions included.
 380198            await VerifyRelationsAsync(connection, transaction, cancellationToken).ConfigureAwait(false);
 199
 378200            await transaction.CommitAsync(cancellationToken).ConfigureAwait(false);
 378201            _created = true;
 378202        }
 203        finally
 204        {
 384205            _ensureGate.Release();
 206        }
 5937207    }
 208
 209    private Task VerifyRelationsAsync(NpgsqlConnection connection, NpgsqlTransaction? transaction, CancellationToken can
 383210        => PostgreSqlRelationVerifier.VerifyAsync(
 383211            connection,
 383212            transaction,
 383213            _options.SchemaName,
 383214            "channel",
 383215            [
 383216                new(_options.RecoveryStateTable, 'r', Columns:
 383217                    [
 383218                        new("correlation_id", "text", Nullable: false, RequiresDeterministicCollation: true),
 383219                        new("registration_id", "uuid", Nullable: false, RequiresDeterministicCollation: true),
 383220                        new("state_json", "text", Nullable: false),
 383221                        new("expires_at", "timestamp with time zone", Nullable: false),
 383222                        new("registered_at", "timestamp with time zone", Nullable: false, DefaultExpression: "now()"),
 383223                    ], PrimaryKey: ["correlation_id", "registration_id"]),
 383224                new(_options.MessageTable, 'r', Columns:
 383225                    [
 383226                        new("id", "uuid", Nullable: false),
 383227                        new("correlation_id", "text", Nullable: false, RequiresDeterministicCollation: true),
 383228                        new("envelope_json", "text", Nullable: false),
 383229                        new("created_at", "timestamp with time zone", Nullable: false, DefaultExpression: "now()"),
 383230                        new("expires_at", "timestamp with time zone", Nullable: false),
 383231                        new("acked_at", "timestamp with time zone", Nullable: true),
 383232                        new("acked_seq", "bigint", Nullable: true),
 383233                        new("recovery_claimed", "boolean", Nullable: false, DefaultExpression: "false"),
 383234                    ], PrimaryKey: ["id"]),
 383235                new(_options.SubscriberTable, 'r', Columns:
 383236                    [
 383237                        new("correlation_id", "text", Nullable: false, RequiresDeterministicCollation: true),
 383238                        new("registration_id", "uuid", Nullable: false, RequiresDeterministicCollation: true),
 383239                        new("instance_id", "text", Nullable: false),
 383240                        new("expires_at", "timestamp with time zone", Nullable: false),
 383241                    ], PrimaryKey: ["correlation_id", "registration_id"]),
 383242                new(AckSequenceName, 'S'),
 383243                new(IndexName(_options.RecoveryStateTable, "expires"), 'i', _options.RecoveryStateTable, ["expires_at"])
 383244                new(IndexName(_options.MessageTable, "correlation_created"), 'i', _options.MessageTable, ["correlation_i
 383245                new(IndexName(_options.MessageTable, "expires"), 'i', _options.MessageTable, ["expires_at"]),
 383246                new(IndexName(_options.SubscriberTable, "expires"), 'i', _options.SubscriberTable, ["expires_at"]),
 383247            ],
 383248            cancellationToken);
 249
 250
 251    private async Task ValidateManagedSchemaAsync(CancellationToken cancellationToken)
 252    {
 54253        await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 254        try
 255        {
 54256            if (_created)
 0257                return;
 258
 54259            await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 260            bool hasColumn;
 261            bool hasSequence;
 262            // The probe's command and reader are scoped so they are disposed before the relation
 263            // verification below reuses this connection — Npgsql allows one command in progress.
 9264            await using (var command = connection.CreateCommand())
 265            {
 266                // relkind = 'S' precisely: to_regclass matches ANY relation, so a table sharing the
 267                // sequence's name (the pre-fix truncation collision) passed validation and failed at
 268                // the first nextval instead.
 9269                command.CommandText =
 9270                    """
 9271                    SELECT
 9272                      EXISTS (SELECT 1 FROM information_schema.columns
 9273                              WHERE table_schema = @schema AND table_name = @table AND column_name = 'acked_seq'),
 9274                      EXISTS (SELECT 1 FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace
 9275                              WHERE n.nspname = @schema AND c.relname = @sequence AND c.relkind = 'S');
 9276                    """;
 9277                command.Parameters.AddWithValue("schema", _options.SchemaName);
 9278                command.Parameters.AddWithValue("table", _options.MessageTable);
 9279                command.Parameters.AddWithValue("sequence", AckSequenceName);
 9280                await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 9281                await reader.ReadAsync(cancellationToken).ConfigureAwait(false);
 9282                hasColumn = reader.GetBoolean(0);
 9283                hasSequence = reader.GetBoolean(1);
 9284            }
 9285            if (!hasColumn || !hasSequence)
 286            {
 6287                throw new InvalidOperationException(
 6288                    $"The PostgreSQL channel schema is managed manually (AutoCreateSchema = false) but is missing " +
 6289                    $"objects this version requires: " +
 6290                    $"{(hasColumn ? "" : $"column {MessageTable}.acked_seq")}{(!hasColumn && !hasSequence ? " and " : ""
 6291                    $"Apply the migration and restart: " +
 6292                    $"ALTER TABLE {MessageTable} ADD COLUMN IF NOT EXISTS acked_seq bigint NULL; " +
 6293                    $"CREATE SEQUENCE IF NOT EXISTS {AckSequence} AS bigint; " +
 6294                    "See docs/postgresql.md, section 'Upgrading a manually managed schema'.");
 295            }
 296
 297            // Full relation verification on the managed path too (transport/flow-store parity):
 298            // an operator-provisioned table with the wrong shape — a nondeterministic
 299            // correlation_id collation above all — previously passed startup here and
 300            // misrouted silently at runtime, which is exactly what verification exists to catch.
 3301            await VerifyRelationsAsync(connection, transaction: null, cancellationToken).ConfigureAwait(false);
 302
 2303            _created = true;
 2304        }
 305        finally
 306        {
 54307            _ensureGate.Release();
 308        }
 2309    }
 310
 311    public async Task SaveRecoveryStateAsync(string correlationId, RecoveryState state, TimeSpan ttl, CancellationToken 
 312    {
 425313        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 425314        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 425315        await using var command = connection.CreateCommand();
 425316        command.CommandText =
 425317            $"""
 425318            INSERT INTO {RecoveryTable} (correlation_id, registration_id, state_json, expires_at, registered_at)
 425319            VALUES (@correlation_id, @registration_id, @state_json, now() + @ttl, now())
 425320            ON CONFLICT (correlation_id, registration_id)
 425321            DO UPDATE SET state_json = EXCLUDED.state_json,
 425322                          expires_at = EXCLUDED.expires_at,
 425323                          registered_at = EXCLUDED.registered_at;
 425324            """;
 425325        command.Parameters.AddWithValue("correlation_id", correlationId);
 425326        command.Parameters.AddWithValue("registration_id", state.RegistrationId);
 425327        command.Parameters.Add("state_json", NpgsqlDbType.Text).Value = AsyncResponseJson.Serialize(state);
 425328        command.Parameters.AddWithValue("ttl", ttl);
 425329        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 425330    }
 331
 332    public async Task<IReadOnlyList<string>> LoadRecoveryStatesAsync(string correlationId, CancellationToken cancellatio
 333    {
 37334        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 37335        if (ShouldPrune(ref _lastRecoveryPruneTicks))
 37336            await PruneExpiredRecoveryAsync(correlationId, cancellationToken).ConfigureAwait(false);
 337
 37338        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 37339        await using var command = connection.CreateCommand();
 37340        command.CommandText =
 37341            $"""
 37342            SELECT state_json::text
 37343            FROM {RecoveryTable}
 37344            WHERE correlation_id = @correlation_id AND expires_at > now()
 37345            ORDER BY registered_at;
 37346            """;
 37347        command.Parameters.AddWithValue("correlation_id", correlationId);
 348
 37349        var states = new List<string>();
 37350        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 62351        while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 25352            states.Add(reader.GetString(0));
 37353        return states;
 37354    }
 355
 356    public async Task<bool> DeleteRecoveryStateAsync(string correlationId, Guid registrationId, CancellationToken cancel
 357    {
 427358        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 427359        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 427360        await using var command = connection.CreateCommand();
 427361        command.CommandText = $"DELETE FROM {RecoveryTable} WHERE correlation_id = @correlation_id AND registration_id =
 427362        command.Parameters.AddWithValue("correlation_id", correlationId);
 427363        command.Parameters.AddWithValue("registration_id", registrationId);
 427364        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 427365    }
 366
 367    public async IAsyncEnumerable<string> ScanRecoveryStateJsonAsync([System.Runtime.CompilerServices.EnumeratorCancella
 368    {
 1369        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 1370        await PruneExpiredRecoveryAsync(null, cancellationToken).ConfigureAwait(false);
 371
 1372        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1373        await using var command = connection.CreateCommand();
 1374        command.CommandText =
 1375            $"""
 1376            SELECT state_json::text
 1377            FROM {RecoveryTable}
 1378            WHERE expires_at > now()
 1379            ORDER BY registered_at;
 1380            """;
 1381        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 2382        while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 1383            yield return reader.GetString(0);
 1384    }
 385
 386    /// <summary>
 387    /// Inserts a response envelope row and notifies listeners. The caller supplies the message id so
 388    /// the insert is idempotent under retry (<c>ON CONFLICT DO NOTHING</c>); the NOTIFY still fires so
 389    /// a retried publish never strands an active waiter. Returns the same-process fast-path
 390    /// message carrying the row's server-stamped <c>created_at</c> — and, on a duplicate, the
 391    /// ORIGINAL row's settlement columns, so the fast path compares against subscription
 392    /// watermarks exactly as the sweep does (a fabricated null <c>acked_at</c> replayed an
 393    /// already-consumed response to a waiter registered after the ack).
 394    /// </summary>
 395    public Task<PostgreSqlChannelMessage> InsertMessageAsync(Guid id, string correlationId, string envelopeJson, TimeSpa
 600396        => AsyncResponseRetry.ExecuteAsync(
 600397            token => InsertMessageOnceAsync(id, correlationId, envelopeJson, retention, token),
 600398            IsTransient,
 600399            _options.PublishMaxAttempts,
 600400            _options.PublishRetryBaseDelay,
 600401            _options.PublishRetryMaxDelay,
 600402            cancellationToken);
 403
 404    private async Task<PostgreSqlChannelMessage> InsertMessageOnceAsync(Guid id, string correlationId, string envelopeJs
 405    {
 600406        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 600407        if (ShouldPrune(ref _lastMessagePruneTicks))
 600408            await PruneExpiredMessagesAsync(cancellationToken).ConfigureAwait(false);
 409
 600410        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 600411        await using var command = connection.CreateCommand();
 412        // Single statement: the final SELECT both fires the NOTIFY exactly once and returns the
 413        // fresh row's server-stamped created_at via RETURNING — NULL when the idempotent insert
 414        // hit a duplicate, which the separate lookup below resolves.
 600415        command.CommandText =
 600416            $"""
 600417            WITH inserted AS (
 600418                INSERT INTO {MessageTable} (id, correlation_id, envelope_json, expires_at)
 600419                VALUES (@id, @correlation_id, @envelope_json, now() + @retention)
 600420                ON CONFLICT (id) DO NOTHING
 600421                RETURNING created_at
 600422            )
 600423            SELECT (SELECT created_at FROM inserted) AS created_at,
 600424                   pg_notify(@channel, @payload);
 600425            """;
 600426        command.Parameters.AddWithValue("id", id);
 600427        command.Parameters.AddWithValue("correlation_id", correlationId);
 600428        command.Parameters.Add("envelope_json", NpgsqlDbType.Text).Value = envelopeJson;
 600429        command.Parameters.AddWithValue("retention", retention);
 600430        command.Parameters.AddWithValue("channel", NotificationChannel);
 600431        command.Parameters.AddWithValue("payload", NotifyPayload(correlationId));
 432        DateTimeOffset? createdAt;
 600433        await using (var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false))
 434        {
 600435            await reader.ReadAsync(cancellationToken).ConfigureAwait(false);
 600436            createdAt = reader.IsDBNull(0) ? null : reader.GetFieldValue<DateTimeOffset>(0);
 437        }
 438
 600439        if (createdAt is { } stamped)
 596440            return new PostgreSqlChannelMessage(id, correlationId, envelopeJson, stamped);
 441
 442        // Duplicate: a publish retry, or a CONCURRENT idempotent publish (ON CONFLICT detects the
 443        // other transaction's row against latest data, while a same-statement subquery would read
 444        // under this statement's older snapshot — reproduced on PostgreSQL 16). A fresh statement
 445        // gets a fresh read-committed snapshot and resolves both deterministically, and it reads
 446        // the original row's settlement columns for the fast-path watermark.
 4447        await using var lookup = connection.CreateCommand();
 4448        lookup.CommandText = $"SELECT created_at, acked_at, acked_seq FROM {MessageTable} WHERE id = @id;";
 4449        lookup.Parameters.AddWithValue("id", id);
 4450        await using var existing = await lookup.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 4451        if (await existing.ReadAsync(cancellationToken).ConfigureAwait(false))
 452        {
 4453            return new PostgreSqlChannelMessage(
 4454                id,
 4455                correlationId,
 4456                envelopeJson,
 4457                existing.GetFieldValue<DateTimeOffset>(0),
 4458                existing.IsDBNull(1) ? null : existing.GetFieldValue<DateTimeOffset>(1),
 4459                existing.IsDBNull(2) ? null : existing.GetInt64(2));
 460        }
 461
 462        // Only reachable when the duplicate's original row is genuinely gone (pruned
 463        // mid-publish): the message is not persisted, and reporting success with a fabricated
 464        // app-clock timestamp would both lie about persistence and feed a client clock into
 465        // the server-clock watermark.
 0466        throw new InvalidOperationException(
 0467            $"PostgreSQL response insert for message {id} found no row after a duplicate: the original no longer exists 
 600468    }
 469
 470    public async Task<IReadOnlyList<PostgreSqlChannelMessage>> LoadMessagesAsync(
 471        string correlationId,
 472        DateTimeOffset sinceUtc,
 473        int batchSize,
 474        DateTimeOffset? afterCreatedAtUtc,
 475        Guid? afterId,
 476        CancellationToken cancellationToken)
 477    {
 1037478        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 1027479        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 998480        await using var command = connection.CreateCommand();
 481        // The envelope travels only for rows nobody has acknowledged yet. Acknowledged rows are
 482        // the consumed history the sweep re-reads on every tick (they stay in the result set so a
 483        // fan-out waiter in ANOTHER process still receives them): shipping their bodies with each
 484        // sweep made a long-lived progress subscription's cost grow with its whole retained
 485        // history. The shared sweep fetches the envelope by id for the rare acknowledged row a
 486        // live subscription has not seen.
 998487        command.CommandText =
 998488            $"""
 998489            SELECT id, correlation_id, CASE WHEN acked_at IS NULL THEN envelope_json::text END, created_at, acked_at, ac
 998490            FROM {MessageTable}
 998491            WHERE correlation_id = @correlation_id
 998492              AND created_at >= @since
 998493              AND expires_at > now()
 998494              {(afterCreatedAtUtc is null ? "" : "AND (created_at > @after_created_at OR (created_at = @after_created_at
 998495            ORDER BY created_at, id
 998496            LIMIT @limit;
 998497            """;
 998498        command.Parameters.AddWithValue("correlation_id", correlationId);
 998499        command.Parameters.AddWithValue("since", sinceUtc);
 998500        command.Parameters.AddWithValue("limit", batchSize);
 998501        if (afterCreatedAtUtc is not null)
 502        {
 233503            command.Parameters.AddWithValue("after_created_at", afterCreatedAtUtc.Value);
 233504            command.Parameters.AddWithValue("after_id", afterId ?? throw new ArgumentNullException(nameof(afterId)));
 505        }
 506
 998507        return await ReadMessagesAsync(command, batchSize, cancellationToken).ConfigureAwait(false);
 989508    }
 509
 510    /// <summary>
 511    /// The full rows (envelope included) for <paramref name="ids"/> under
 512    /// <paramref name="correlationId"/>, in sweep order — how the dispatch sweep hydrates the
 513    /// header-only acknowledged rows it still has to deliver. A row pruned between the sweep's
 514    /// page and this read is simply absent.
 515    /// </summary>
 516    public async Task<IReadOnlyList<PostgreSqlChannelMessage>> LoadMessagesByIdAsync(
 517        string correlationId,
 518        IReadOnlyList<Guid> ids,
 519        CancellationToken cancellationToken)
 520    {
 10521        if (ids.Count == 0)
 2522            return [];
 523
 8524        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 8525        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 6526        await using var command = connection.CreateCommand();
 6527        command.CommandText =
 6528            $"""
 6529            SELECT id, correlation_id, envelope_json::text, created_at, acked_at, acked_seq
 6530            FROM {MessageTable}
 6531            WHERE correlation_id = @correlation_id
 6532              AND id = ANY(@ids)
 6533              AND expires_at > now()
 6534            ORDER BY created_at, id;
 6535            """;
 6536        command.Parameters.AddWithValue("correlation_id", correlationId);
 6537        command.Parameters.AddWithValue("ids", ids is Guid[] array ? array : [.. ids]);
 6538        return await ReadMessagesAsync(command, ids.Count, cancellationToken).ConfigureAwait(false);
 8539    }
 540
 541    private static async Task<IReadOnlyList<PostgreSqlChannelMessage>> ReadMessagesAsync(NpgsqlCommand command, int capa
 542    {
 1004543        var messages = new List<PostgreSqlChannelMessage>(capacity);
 1004544        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 1760545        while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 765546            messages.Add(new PostgreSqlChannelMessage(
 765547                reader.GetGuid(0),
 765548                reader.GetString(1),
 765549                reader.IsDBNull(2) ? null : reader.GetString(2),
 765550                reader.GetFieldValue<DateTimeOffset>(3),
 765551                reader.IsDBNull(4) ? null : reader.GetFieldValue<DateTimeOffset>(4),
 765552                reader.IsDBNull(5) ? null : reader.GetInt64(5)));
 995553        return messages;
 995554    }
 555
 556    /// <summary>
 557    /// Atomically claims a message for live delivery: sets <c>acked_at</c> unless the publisher has
 558    /// already routed it to the lost-subscriber path (<c>recovery_claimed</c>). Returns <c>false</c>
 559    /// when recovery owns the message, so a slow-but-live waiter does not deliver a response the
 560    /// recovery callback already handled. Multiple processes may each win this claim, preserving
 561    /// cross-process fan-out, because it gates only on <c>recovery_claimed</c>, not on <c>acked_at</c>.
 562    /// </summary>
 563    public async Task<bool> TryClaimForDeliveryAsync(Guid messageId, CancellationToken cancellationToken)
 564    {
 525565        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 519566        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 519567        await using var command = connection.CreateCommand();
 568        // The sequence is stamped ONLY when this same update transitions acked_at from null (SET
 569        // expressions read the pre-update row): a row acked by a pre-sequence build must stay
 570        // permanently unsequenced. Back-filling it on a later fan-out re-claim would pair an OLD
 571        // acked_at with a FRESH sequence value, and a waiter that registered in the original ack's
 572        // tick would then read the tie as post-registration fan-out — replaying a response its
 573        // predecessor consumed.
 519574        command.CommandText =
 519575            $"""
 519576            UPDATE {MessageTable}
 519577            SET acked_at = COALESCE(acked_at, now()),
 519578                acked_seq = CASE WHEN acked_at IS NULL THEN nextval('{AckSequence}') ELSE acked_seq END
 519579            WHERE id = @id AND NOT recovery_claimed AND expires_at > now()
 519580            RETURNING id;
 519581            """;
 519582        command.Parameters.AddWithValue("id", messageId);
 519583        var result = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
 519584        return result is not null and not DBNull;
 519585    }
 586
 587    /// <summary>
 588    /// Atomically claims a message for the lost-subscriber path: sets <c>recovery_claimed</c> only
 589    /// while no waiter has delivered (<c>acked_at IS NULL</c>). Returns <c>true</c> when recovery wins;
 590    /// <c>false</c> means a live waiter already took the message, so the publisher must not also fire
 591    /// the recovery callback. Row-level locking serializes this against <see cref="TryClaimForDeliveryAsync"/>.
 592    /// </summary>
 593    public async Task<bool> TryClaimForRecoveryAsync(Guid messageId, CancellationToken cancellationToken)
 594    {
 7595        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 7596        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 7597        await using var command = connection.CreateCommand();
 7598        command.CommandText =
 7599            $"""
 7600            UPDATE {MessageTable}
 7601            SET recovery_claimed = true
 7602            WHERE id = @id AND acked_at IS NULL
 7603            RETURNING id;
 7604            """;
 7605        command.Parameters.AddWithValue("id", messageId);
 7606        var result = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
 7607        return result is not null and not DBNull;
 7608    }
 609
 610    /// <summary>
 611    /// One round trip for a subscription's registration watermark: the server's UTC clock (for
 612    /// the created-at bound) and a fresh position in the monotonic ack sequence (for the exact
 613    /// acked-history bound — see the watermark in the shared channel base).
 614    /// </summary>
 615    public async Task<(DateTimeOffset ServerTimeUtc, long StartSeq)> GetSubscriptionStartAsync(CancellationToken cancell
 616    {
 415617        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 413618        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 413619        await using var command = connection.CreateCommand();
 413620        command.CommandText = $"SELECT now(), nextval('{AckSequence}');";
 413621        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 413622        await reader.ReadAsync(cancellationToken).ConfigureAwait(false);
 413623        return (reader.GetFieldValue<DateTimeOffset>(0).ToUniversalTime(), reader.GetInt64(1));
 413624    }
 625
 626    /// <summary>Returns the database server's current UTC time, used as a clock-safe delivery watermark.</summary>
 627    public async Task<DateTimeOffset> GetServerTimeUtcAsync(CancellationToken cancellationToken)
 628    {
 5629        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 5630        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 3631        await using var command = connection.CreateCommand();
 3632        command.CommandText = "SELECT now();";
 3633        var result = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
 3634        return result switch
 3635        {
 0636            DateTimeOffset dto => dto.ToUniversalTime(),
 3637            DateTime dt => new DateTimeOffset(DateTime.SpecifyKind(dt, DateTimeKind.Utc), TimeSpan.Zero),
 0638            _ => DateTimeOffset.UtcNow
 3639        };
 3640    }
 641
 642    public async Task<bool> IsMessageAcknowledgedAsync(Guid messageId, CancellationToken cancellationToken)
 643    {
 160644        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 158645        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 156646        await using var command = connection.CreateCommand();
 156647        command.CommandText = $"SELECT acked_at IS NOT NULL FROM {MessageTable} WHERE id = @id AND expires_at > now();";
 156648        command.Parameters.AddWithValue("id", messageId);
 156649        var result = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
 156650        return result is bool acknowledged && acknowledged;
 156651    }
 652
 653    public async Task UpsertSubscriberAsync(string correlationId, Guid registrationId, string instanceId, TimeSpan ttl, 
 654    {
 417655        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 417656        if (ShouldPrune(ref _lastSubscriberPruneTicks))
 417657            await PruneExpiredSubscribersAsync(correlationId, cancellationToken).ConfigureAwait(false);
 658
 417659        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 417660        await using var command = connection.CreateCommand();
 417661        command.CommandText =
 417662            $"""
 417663            INSERT INTO {SubscriberTable} (correlation_id, registration_id, instance_id, expires_at)
 417664            VALUES (@correlation_id, @registration_id, @instance_id, now() + @ttl)
 417665            ON CONFLICT (correlation_id, registration_id)
 417666            DO UPDATE SET instance_id = EXCLUDED.instance_id,
 417667                          expires_at = EXCLUDED.expires_at;
 417668            """;
 417669        command.Parameters.AddWithValue("correlation_id", correlationId);
 417670        command.Parameters.AddWithValue("registration_id", registrationId);
 417671        command.Parameters.AddWithValue("instance_id", instanceId);
 417672        command.Parameters.AddWithValue("ttl", ttl);
 417673        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 417674    }
 675
 676    public async Task HeartbeatSubscribersAsync(
 677        string instanceId,
 678        IReadOnlyCollection<(string CorrelationId, Guid RegistrationId)> registrations,
 679        TimeSpan ttl,
 680        CancellationToken cancellationToken)
 681    {
 117682        if (registrations.Count == 0)
 1683            return;
 684
 116685        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 109686        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 107687        await using var command = connection.CreateCommand();
 688
 689        // UPSERT rather than a bare UPDATE: the caller only heartbeats registrations that are live
 690        // in this process, so a missing row means the pruner deleted it (e.g. after a >timeout
 691        // stall) — re-creating it here is what brings the waiter back from "permanently invisible".
 107692        var correlationIds = new string[registrations.Count];
 107693        var registrationIds = new Guid[registrations.Count];
 107694        var index = 0;
 522695        foreach (var (correlationId, registrationId) in registrations)
 696        {
 154697            correlationIds[index] = correlationId;
 154698            registrationIds[index] = registrationId;
 154699            index++;
 700        }
 701
 107702        command.CommandText =
 107703            $"""
 107704            INSERT INTO {SubscriberTable} (correlation_id, registration_id, instance_id, expires_at)
 107705            SELECT correlation_id, registration_id, @instance_id, now() + @ttl
 107706            FROM unnest(@correlation_ids, @registration_ids) AS live (correlation_id, registration_id)
 107707            ON CONFLICT (correlation_id, registration_id)
 107708            DO UPDATE SET instance_id = EXCLUDED.instance_id,
 107709                          expires_at = EXCLUDED.expires_at;
 107710            """;
 107711        command.Parameters.AddWithValue("instance_id", instanceId);
 107712        command.Parameters.AddWithValue("correlation_ids", NpgsqlDbType.Array | NpgsqlDbType.Text, correlationIds);
 107713        command.Parameters.AddWithValue("registration_ids", NpgsqlDbType.Array | NpgsqlDbType.Uuid, registrationIds);
 107714        command.Parameters.AddWithValue("ttl", ttl);
 107715        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 108716    }
 717
 718    public async Task DeleteSubscriberAsync(string correlationId, Guid registrationId, CancellationToken cancellationTok
 719    {
 441720        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 431721        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 431722        await using var command = connection.CreateCommand();
 431723        command.CommandText = $"DELETE FROM {SubscriberTable} WHERE correlation_id = @correlation_id AND registration_id
 431724        command.Parameters.AddWithValue("correlation_id", correlationId);
 431725        command.Parameters.AddWithValue("registration_id", registrationId);
 431726        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 430727    }
 728
 729    public async Task<long> CountActiveSubscribersAsync(string correlationId, CancellationToken cancellationToken)
 730    {
 579731        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 569732        if (ShouldPrune(ref _lastSubscriberPruneTicks))
 569733            await PruneExpiredSubscribersAsync(correlationId, cancellationToken).ConfigureAwait(false);
 734
 567735        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 567736        await using var command = connection.CreateCommand();
 567737        command.CommandText =
 567738            $"""
 567739            SELECT count(*)::bigint
 567740            FROM {SubscriberTable}
 567741            WHERE correlation_id = @correlation_id AND expires_at > now();
 567742            """;
 567743        command.Parameters.AddWithValue("correlation_id", correlationId);
 567744        var result = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
 567745        return result is long count ? count : 0L;
 567746    }
 747
 748    public async Task ExecuteListenAsync(Func<string?, Task> onNotification, CancellationToken cancellationToken)
 749    {
 365750        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 361751        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 870752        connection.Notification += (_, args) => _ = onNotification(args.Payload);
 359753        await using (var command = connection.CreateCommand())
 754        {
 359755            command.CommandText = $"LISTEN {Quote(NotificationChannel)};";
 359756            await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 757        }
 758
 870759        while (!cancellationToken.IsCancellationRequested)
 870760            await connection.WaitAsync(cancellationToken).ConfigureAwait(false);
 0761    }
 762
 763    private async Task PruneExpiredRecoveryAsync(string? correlationId, CancellationToken cancellationToken)
 764    {
 38765        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 38766        await using var command = connection.CreateCommand();
 38767        command.CommandText = correlationId is null
 38768            ? $"DELETE FROM {RecoveryTable} WHERE expires_at <= now();"
 38769            : $"DELETE FROM {RecoveryTable} WHERE correlation_id = @correlation_id AND expires_at <= now();";
 38770        if (correlationId is not null)
 37771            command.Parameters.AddWithValue("correlation_id", correlationId);
 38772        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 38773    }
 774
 775    private async Task PruneExpiredMessagesAsync(CancellationToken cancellationToken)
 776    {
 600777        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 600778        await using var command = connection.CreateCommand();
 600779        command.CommandText = $"DELETE FROM {MessageTable} WHERE expires_at <= now();";
 600780        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 600781    }
 782
 783    private async Task PruneExpiredSubscribersAsync(string? correlationId, CancellationToken cancellationToken)
 784    {
 988785        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 984786        await using var command = connection.CreateCommand();
 984787        command.CommandText = correlationId is null
 984788            ? $"DELETE FROM {SubscriberTable} WHERE expires_at <= now();"
 984789            : $"DELETE FROM {SubscriberTable} WHERE correlation_id = @correlation_id AND expires_at <= now();";
 984790        if (correlationId is not null)
 984791            command.Parameters.AddWithValue("correlation_id", correlationId);
 984792        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 984793    }
 794
 795    public static void ValidateIdentifier(string? value, string name)
 796    {
 4564797        if (string.IsNullOrWhiteSpace(value))
 4798            throw new InvalidOperationException($"{nameof(PostgreSqlAsyncResponseChannelOptions)}.{name} must be configu
 4560799        if (!IsIdentifier(value))
 6800            throw new InvalidOperationException(
 6801                $"{nameof(PostgreSqlAsyncResponseChannelOptions)}.{name} '{value}' must be a simple PostgreSQL identifie
 802        // PostgreSQL TRUNCATES over-limit identifiers silently (a NOTICE, not an error), so an
 803        // over-limit configured name would create/address an object under a different name.
 4554804        if (value.Length > IdentifierCap)
 4805            throw new InvalidOperationException(
 4806                $"{nameof(PostgreSqlAsyncResponseChannelOptions)}.{name} '{value}' is {value.Length} characters; Postgre
 4550807    }
 808
 809    private static bool IsIdentifier(string value)
 810    {
 4568811        if (value.Length == 0 || !(char.IsAsciiLetter(value[0]) || value[0] == '_'))
 8812            return false;
 813
 255756814        foreach (var c in value)
 815        {
 123320816            if (!(char.IsAsciiLetterOrDigit(c) || c == '_'))
 4817                return false;
 818        }
 819
 4556820        return true;
 821    }
 822
 4107823    private static string Quote(string identifier) => "\"" + identifier + "\"";
 824
 825    /// <summary>
 826    /// A single-quoted SQL string literal, for the catalog lookups inside the DDL's DO block where
 827    /// a parameter cannot be bound. Names are already validated by the options; the doubling keeps
 828    /// the literal well-formed regardless.
 829    /// </summary>
 1528830    private static string SqlLiteral(string value) => "'" + value.Replace("'", "''", StringComparison.Ordinal) + "'";
 831
 832    /// <summary>PostgreSQL's identifier length cap (NAMEDATALEN - 1); longer names are silently truncated server-side.<
 833    internal const int IdentifierCap = 63;
 834
 835    // Suffix space is RESERVED before capping in BOTH derived-name helpers (see
 836    // RelationalNamePlan.DerivedName, the shared implementation): truncating the whole
 837    // "{table}{suffix}" let a maximum-length table name produce exactly the table's own name — the
 838    // derived object then collided with the table (indexes, sequences, and tables share one
 839    // relation namespace), CREATE ... IF NOT EXISTS silently skipped it, and the store ran with a
 840    // missing sequence (runtime failure) or missing indexes (silent full scans).
 841    private static string IndexName(string table, string suffix)
 6710842        => RelationalNamePlan.DerivedName(table, $"_{suffix}_idx", IdentifierCap);
 843
 844    private static string SequenceName(string table)
 1356845        => RelationalNamePlan.DerivedName(table, "_ack_seq", IdentifierCap);
 846
 847    /// <summary>
 848    /// Validates the complete effective object-name plan — configured tables plus every derived
 849    /// index and sequence name — for pairwise distinctness. Suffix reservation makes one table's
 850    /// derived names collision-free, but two long tables whose reserved stems truncate identically
 851    /// still derive the same index name, and a configured table can occupy a derived name outright;
 852    /// either way <c>CREATE ... IF NOT EXISTS</c> silently skips the object. Comparison is
 853    /// case-insensitive: the DDL quotes identifiers (case-sensitive to PostgreSQL), but a plan
 854    /// distinct only by letter case is a misconfiguration magnet and is rejected for parity with
 855    /// SQL Server's case-insensitive catalogs.
 856    /// </summary>
 857    public static void ValidateNamePlan(PostgreSqlAsyncResponseChannelOptions options)
 858    {
 912859        (string Role, string Name)[] plan =
 912860        [
 912861            ($"{nameof(options.RecoveryStateTable)} table", options.RecoveryStateTable),
 912862            ($"{nameof(options.MessageTable)} table", options.MessageTable),
 912863            ($"{nameof(options.SubscriberTable)} table", options.SubscriberTable),
 912864            ("ack sequence (derived from MessageTable)", SequenceName(options.MessageTable)),
 912865            ("RecoveryStateTable expiry index", IndexName(options.RecoveryStateTable, "expires")),
 912866            ("MessageTable correlation index", IndexName(options.MessageTable, "correlation_created")),
 912867            ("MessageTable expiry index", IndexName(options.MessageTable, "expires")),
 912868            ("SubscriberTable expiry index", IndexName(options.SubscriberTable, "expires")),
 912869        ];
 912870        RelationalNamePlan.RequireDistinct(
 912871            plan,
 912872            nameof(PostgreSqlAsyncResponseChannelOptions),
 912873            ". All tables and the index/sequence names derived from them share one namespace and must be distinct " +
 912874            "(long names reserve suffix space by truncating the table stem, which can make distinct tables derive " +
 912875            "the same name). Shorten or de-overlap the configured table names.");
 902876    }
 877
 878    /// <summary>NOTIFY payload for a publish: the correlation id, or empty when it is too long to carry.</summary>
 879    private static string NotifyPayload(string correlationId)
 604880        => Encoding.UTF8.GetByteCount(correlationId) <= MaxNotifyPayloadBytes ? correlationId : string.Empty;
 881
 28882    internal static bool IsTransient(Exception exception) => PostgreSqlTransientFaults.IsTransient(exception);
 883
 884    /// <summary>
 885    /// Stable 64-bit advisory-lock key for serializing schema creation. Uses FNV-1a over a
 886    /// schema-scoped discriminator: it must be deterministic across processes (so
 887    /// <see cref="string.GetHashCode()"/>, which is per-process randomized, is unusable) and identical
 888    /// to the transport store's key for the same schema so both serialize their shared CREATE SCHEMA.
 889    /// </summary>
 890    internal static long SchemaAdvisoryLockKey(string schemaName)
 891    {
 892        const ulong offset = 14695981039346656037UL;
 893        const ulong prime = 1099511628211UL;
 455894        var hash = offset;
 33838895        foreach (var b in Encoding.UTF8.GetBytes($"asyncresponse:ddl:{schemaName}"))
 896        {
 16464897            hash ^= b;
 16464898            hash *= prime;
 899        }
 900
 455901        return unchecked((long)hash);
 902    }
 903
 904    /// <summary>
 905    /// Time-gates opportunistic pruning so the housekeeping DELETE runs at most once per
 906    /// <see cref="PostgreSqlAsyncResponseChannelOptions.PruneInterval"/> instead of on every operation.
 907    /// Read queries already filter on <c>expires_at</c>, so throttling pruning never affects correctness.
 908    /// </summary>
 909    private bool ShouldPrune(ref long lastTicks)
 910    {
 1629911        var interval = _options.PruneInterval;
 1629912        if (interval <= TimeSpan.Zero)
 1623913            return true;
 914
 6915        var now = DateTime.UtcNow.Ticks;
 6916        var last = Interlocked.Read(ref lastTicks);
 6917        return now - last >= interval.Ticks
 6918            && Interlocked.CompareExchange(ref lastTicks, now, last) == last;
 919    }
 920}
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.Channels.PostgreSQL.PostgreSqlAsyncResponseChannelOptions>)
get_Schema()
get_RecoveryTable()
get_MessageTable()
get_SubscriberTable()
get_AckSequence()
get_AckSequenceName()
get_NotificationChannel()
EnsureCreatedAsync()
VerifyRelationsAsync(Npgsql.NpgsqlConnection,Npgsql.NpgsqlTransaction,System.Threading.CancellationToken)
ValidateManagedSchemaAsync()
SaveRecoveryStateAsync()
LoadRecoveryStatesAsync()
DeleteRecoveryStateAsync()
ScanRecoveryStateJsonAsync()
InsertMessageAsync(System.Guid,System.String,System.String,System.TimeSpan,System.Threading.CancellationToken)
InsertMessageOnceAsync()
LoadMessagesAsync()
LoadMessagesByIdAsync()
ReadMessagesAsync()
TryClaimForDeliveryAsync()
TryClaimForRecoveryAsync()
GetSubscriptionStartAsync()
GetServerTimeUtcAsync()
IsMessageAcknowledgedAsync()
UpsertSubscriberAsync()
HeartbeatSubscribersAsync()
DeleteSubscriberAsync()
CountActiveSubscribersAsync()
ExecuteListenAsync()
PruneExpiredRecoveryAsync()
PruneExpiredMessagesAsync()
PruneExpiredSubscribersAsync()
ValidateIdentifier(System.String,System.String)
IsIdentifier(System.String)
Quote(System.String)
SqlLiteral(System.String)
IndexName(System.String,System.String)
SequenceName(System.String)
ValidateNamePlan(AsyncResponse.Channels.PostgreSQL.PostgreSqlAsyncResponseChannelOptions)
NotifyPayload(System.String)
IsTransient(System.Exception)
SchemaAdvisoryLockKey(System.String)
ShouldPrune(System.Int64&)