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

Information
Class: AsyncResponse.Channels.PostgreSQL.PostgreSqlChannelMessage
Assembly: AsyncResponse.Channels.PostgreSQL
File(s): /_/src/Channels/AsyncResponse.Channels.PostgreSQL/PostgreSqlChannelSql.cs
Line coverage
100%
Covered lines: 6
Uncovered lines: 0
Coverable lines: 6
Total lines: 920
Line coverage: 100%
Branch coverage
N/A
Covered branches: 0
Total branches: 0
Branch coverage: N/A
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
get_Id()100%11100%
get_CorrelationId()100%11100%
get_EnvelopeJson()100%11100%
get_CreatedAtUtc()100%11100%
get_AckedAtUtc()100%11100%
get_AckedSeq()100%11100%

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(
 317517    Guid Id,
 171118    string CorrelationId,
 108519    string? EnvelopeJson,
 207520    DateTimeOffset CreatedAtUtc,
 162621    DateTimeOffset? AckedAtUtc = null,
 1322    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;
 33    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
 40    public PostgreSqlChannelSql(NpgsqlDataSource dataSource, Microsoft.Extensions.Options.IOptions<PostgreSqlAsyncRespon
 41    {
 42        _dataSource = dataSource;
 43        _options = options.Value;
 44        _options.Validate();
 45
 46        Schema = Quote(_options.SchemaName);
 47        RecoveryTable = $"{Schema}.{Quote(_options.RecoveryStateTable)}";
 48        MessageTable = $"{Schema}.{Quote(_options.MessageTable)}";
 49        SubscriberTable = $"{Schema}.{Quote(_options.SubscriberTable)}";
 50        AckSequenceName = SequenceName(_options.MessageTable);
 51        AckSequence = $"{Schema}.{Quote(AckSequenceName)}";
 52        _schemaLockKey = SchemaAdvisoryLockKey(_options.SchemaName);
 53    }
 54
 55    public string Schema { get; }
 56    public string RecoveryTable { get; }
 57    public string MessageTable { get; }
 58    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>
 65    public string AckSequence { get; }
 66
 67    /// <summary>Unquoted sequence identifier, for catalog queries.</summary>
 68    public string AckSequenceName { get; }
 69    public string NotificationChannel => _options.NotificationChannel;
 70
 71    public async Task EnsureCreatedAsync(CancellationToken cancellationToken = default)
 72    {
 73        if (_created)
 74            return;
 75
 76        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.
 83            await ValidateManagedSchemaAsync(cancellationToken).ConfigureAwait(false);
 84            return;
 85        }
 86
 87        await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 88        try
 89        {
 90            if (_created)
 91                return;
 92
 93            await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 94            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.
 101            await using (var lockCommand = connection.CreateCommand())
 102            {
 103                lockCommand.Transaction = transaction;
 104                lockCommand.CommandText = "SELECT pg_advisory_xact_lock(@lock_key);";
 105                lockCommand.Parameters.AddWithValue("lock_key", _schemaLockKey);
 106                await lockCommand.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 107            }
 108
 109            await using var command = connection.CreateCommand();
 110            command.Transaction = transaction;
 111            command.CommandText =
 112                $"""
 113                CREATE SCHEMA IF NOT EXISTS {Schema};
 114
 115                CREATE TABLE IF NOT EXISTS {RecoveryTable} (
 116                    correlation_id text NOT NULL,
 117                    registration_id uuid NOT NULL,
 118                    state_json text NOT NULL,
 119                    expires_at timestamptz NOT NULL,
 120                    registered_at timestamptz NOT NULL DEFAULT now(),
 121                    PRIMARY KEY (correlation_id, registration_id)
 122                );
 123                CREATE INDEX IF NOT EXISTS {Quote(IndexName(_options.RecoveryStateTable, "expires"))}
 124                    ON {RecoveryTable} (expires_at);
 125
 126                CREATE TABLE IF NOT EXISTS {MessageTable} (
 127                    id uuid PRIMARY KEY,
 128                    correlation_id text NOT NULL,
 129                    envelope_json text NOT NULL,
 130                    created_at timestamptz NOT NULL DEFAULT now(),
 131                    expires_at timestamptz NOT NULL,
 132                    acked_at timestamptz NULL,
 133                    acked_seq bigint NULL,
 134                    recovery_claimed boolean NOT NULL DEFAULT false
 135                );
 136                ALTER TABLE {MessageTable} ADD COLUMN IF NOT EXISTS recovery_claimed boolean NOT NULL DEFAULT false;
 137                ALTER TABLE {MessageTable} ADD COLUMN IF NOT EXISTS acked_seq bigint NULL;
 138
 139                -- jsonb REJECTS the \u0000 escape that System.Text.Json emits for U+0000 (SQLSTATE
 140                -- 22P05), so any payload, exception message, propagated context value or callback
 141                -- argument containing a NUL was unpublishable on PostgreSQL alone while every other
 142                -- channel delivered it. Nothing here ever queries INSIDE the document — both columns
 143                -- are read back with ::text — so text costs nothing and accepts the whole contract.
 144                -- Guarded so the rewrite happens once, not on every start.
 145                DO $$
 146                BEGIN
 147                    IF EXISTS (
 148                        SELECT 1 FROM information_schema.columns
 149                        WHERE table_schema = {SqlLiteral(_options.SchemaName)} AND table_name = {SqlLiteral(_options.Mes
 150                          AND column_name = 'envelope_json' AND data_type = 'jsonb')
 151                    THEN
 152                        ALTER TABLE {MessageTable} ALTER COLUMN envelope_json TYPE text USING envelope_json::text;
 153                    END IF;
 154
 155                    IF EXISTS (
 156                        SELECT 1 FROM information_schema.columns
 157                        WHERE table_schema = {SqlLiteral(_options.SchemaName)} AND table_name = {SqlLiteral(_options.Rec
 158                          AND column_name = 'state_json' AND data_type = 'jsonb')
 159                    THEN
 160                        ALTER TABLE {RecoveryTable} ALTER COLUMN state_json TYPE text USING state_json::text;
 161                    END IF;
 162                END $$;
 163                CREATE SEQUENCE IF NOT EXISTS {AckSequence} AS bigint;
 164                CREATE INDEX IF NOT EXISTS {Quote(IndexName(_options.MessageTable, "correlation_created"))}
 165                    ON {MessageTable} (correlation_id, created_at);
 166                CREATE INDEX IF NOT EXISTS {Quote(IndexName(_options.MessageTable, "expires"))}
 167                    ON {MessageTable} (expires_at);
 168
 169                CREATE TABLE IF NOT EXISTS {SubscriberTable} (
 170                    correlation_id text NOT NULL,
 171                    registration_id uuid NOT NULL,
 172                    instance_id text NOT NULL,
 173                    expires_at timestamptz NOT NULL,
 174                    PRIMARY KEY (correlation_id, registration_id)
 175                );
 176                CREATE INDEX IF NOT EXISTS {Quote(IndexName(_options.SubscriberTable, "expires"))}
 177                    ON {SubscriberTable} (expires_at);
 178                """;
 179            try
 180            {
 181                await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 182            }
 183            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".
 189                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.
 198            await VerifyRelationsAsync(connection, transaction, cancellationToken).ConfigureAwait(false);
 199
 200            await transaction.CommitAsync(cancellationToken).ConfigureAwait(false);
 201            _created = true;
 202        }
 203        finally
 204        {
 205            _ensureGate.Release();
 206        }
 207    }
 208
 209    private Task VerifyRelationsAsync(NpgsqlConnection connection, NpgsqlTransaction? transaction, CancellationToken can
 210        => PostgreSqlRelationVerifier.VerifyAsync(
 211            connection,
 212            transaction,
 213            _options.SchemaName,
 214            "channel",
 215            [
 216                new(_options.RecoveryStateTable, 'r', Columns:
 217                    [
 218                        new("correlation_id", "text", Nullable: false, RequiresDeterministicCollation: true),
 219                        new("registration_id", "uuid", Nullable: false, RequiresDeterministicCollation: true),
 220                        new("state_json", "text", Nullable: false),
 221                        new("expires_at", "timestamp with time zone", Nullable: false),
 222                        new("registered_at", "timestamp with time zone", Nullable: false, DefaultExpression: "now()"),
 223                    ], PrimaryKey: ["correlation_id", "registration_id"]),
 224                new(_options.MessageTable, 'r', Columns:
 225                    [
 226                        new("id", "uuid", Nullable: false),
 227                        new("correlation_id", "text", Nullable: false, RequiresDeterministicCollation: true),
 228                        new("envelope_json", "text", Nullable: false),
 229                        new("created_at", "timestamp with time zone", Nullable: false, DefaultExpression: "now()"),
 230                        new("expires_at", "timestamp with time zone", Nullable: false),
 231                        new("acked_at", "timestamp with time zone", Nullable: true),
 232                        new("acked_seq", "bigint", Nullable: true),
 233                        new("recovery_claimed", "boolean", Nullable: false, DefaultExpression: "false"),
 234                    ], PrimaryKey: ["id"]),
 235                new(_options.SubscriberTable, 'r', Columns:
 236                    [
 237                        new("correlation_id", "text", Nullable: false, RequiresDeterministicCollation: true),
 238                        new("registration_id", "uuid", Nullable: false, RequiresDeterministicCollation: true),
 239                        new("instance_id", "text", Nullable: false),
 240                        new("expires_at", "timestamp with time zone", Nullable: false),
 241                    ], PrimaryKey: ["correlation_id", "registration_id"]),
 242                new(AckSequenceName, 'S'),
 243                new(IndexName(_options.RecoveryStateTable, "expires"), 'i', _options.RecoveryStateTable, ["expires_at"])
 244                new(IndexName(_options.MessageTable, "correlation_created"), 'i', _options.MessageTable, ["correlation_i
 245                new(IndexName(_options.MessageTable, "expires"), 'i', _options.MessageTable, ["expires_at"]),
 246                new(IndexName(_options.SubscriberTable, "expires"), 'i', _options.SubscriberTable, ["expires_at"]),
 247            ],
 248            cancellationToken);
 249
 250
 251    private async Task ValidateManagedSchemaAsync(CancellationToken cancellationToken)
 252    {
 253        await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 254        try
 255        {
 256            if (_created)
 257                return;
 258
 259            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.
 264            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.
 269                command.CommandText =
 270                    """
 271                    SELECT
 272                      EXISTS (SELECT 1 FROM information_schema.columns
 273                              WHERE table_schema = @schema AND table_name = @table AND column_name = 'acked_seq'),
 274                      EXISTS (SELECT 1 FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace
 275                              WHERE n.nspname = @schema AND c.relname = @sequence AND c.relkind = 'S');
 276                    """;
 277                command.Parameters.AddWithValue("schema", _options.SchemaName);
 278                command.Parameters.AddWithValue("table", _options.MessageTable);
 279                command.Parameters.AddWithValue("sequence", AckSequenceName);
 280                await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 281                await reader.ReadAsync(cancellationToken).ConfigureAwait(false);
 282                hasColumn = reader.GetBoolean(0);
 283                hasSequence = reader.GetBoolean(1);
 284            }
 285            if (!hasColumn || !hasSequence)
 286            {
 287                throw new InvalidOperationException(
 288                    $"The PostgreSQL channel schema is managed manually (AutoCreateSchema = false) but is missing " +
 289                    $"objects this version requires: " +
 290                    $"{(hasColumn ? "" : $"column {MessageTable}.acked_seq")}{(!hasColumn && !hasSequence ? " and " : ""
 291                    $"Apply the migration and restart: " +
 292                    $"ALTER TABLE {MessageTable} ADD COLUMN IF NOT EXISTS acked_seq bigint NULL; " +
 293                    $"CREATE SEQUENCE IF NOT EXISTS {AckSequence} AS bigint; " +
 294                    "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.
 301            await VerifyRelationsAsync(connection, transaction: null, cancellationToken).ConfigureAwait(false);
 302
 303            _created = true;
 304        }
 305        finally
 306        {
 307            _ensureGate.Release();
 308        }
 309    }
 310
 311    public async Task SaveRecoveryStateAsync(string correlationId, RecoveryState state, TimeSpan ttl, CancellationToken 
 312    {
 313        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 314        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 315        await using var command = connection.CreateCommand();
 316        command.CommandText =
 317            $"""
 318            INSERT INTO {RecoveryTable} (correlation_id, registration_id, state_json, expires_at, registered_at)
 319            VALUES (@correlation_id, @registration_id, @state_json, now() + @ttl, now())
 320            ON CONFLICT (correlation_id, registration_id)
 321            DO UPDATE SET state_json = EXCLUDED.state_json,
 322                          expires_at = EXCLUDED.expires_at,
 323                          registered_at = EXCLUDED.registered_at;
 324            """;
 325        command.Parameters.AddWithValue("correlation_id", correlationId);
 326        command.Parameters.AddWithValue("registration_id", state.RegistrationId);
 327        command.Parameters.Add("state_json", NpgsqlDbType.Text).Value = AsyncResponseJson.Serialize(state);
 328        command.Parameters.AddWithValue("ttl", ttl);
 329        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 330    }
 331
 332    public async Task<IReadOnlyList<string>> LoadRecoveryStatesAsync(string correlationId, CancellationToken cancellatio
 333    {
 334        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 335        if (ShouldPrune(ref _lastRecoveryPruneTicks))
 336            await PruneExpiredRecoveryAsync(correlationId, cancellationToken).ConfigureAwait(false);
 337
 338        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 339        await using var command = connection.CreateCommand();
 340        command.CommandText =
 341            $"""
 342            SELECT state_json::text
 343            FROM {RecoveryTable}
 344            WHERE correlation_id = @correlation_id AND expires_at > now()
 345            ORDER BY registered_at;
 346            """;
 347        command.Parameters.AddWithValue("correlation_id", correlationId);
 348
 349        var states = new List<string>();
 350        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 351        while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 352            states.Add(reader.GetString(0));
 353        return states;
 354    }
 355
 356    public async Task<bool> DeleteRecoveryStateAsync(string correlationId, Guid registrationId, CancellationToken cancel
 357    {
 358        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 359        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 360        await using var command = connection.CreateCommand();
 361        command.CommandText = $"DELETE FROM {RecoveryTable} WHERE correlation_id = @correlation_id AND registration_id =
 362        command.Parameters.AddWithValue("correlation_id", correlationId);
 363        command.Parameters.AddWithValue("registration_id", registrationId);
 364        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 365    }
 366
 367    public async IAsyncEnumerable<string> ScanRecoveryStateJsonAsync([System.Runtime.CompilerServices.EnumeratorCancella
 368    {
 369        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 370        await PruneExpiredRecoveryAsync(null, cancellationToken).ConfigureAwait(false);
 371
 372        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 373        await using var command = connection.CreateCommand();
 374        command.CommandText =
 375            $"""
 376            SELECT state_json::text
 377            FROM {RecoveryTable}
 378            WHERE expires_at > now()
 379            ORDER BY registered_at;
 380            """;
 381        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 382        while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 383            yield return reader.GetString(0);
 384    }
 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
 396        => AsyncResponseRetry.ExecuteAsync(
 397            token => InsertMessageOnceAsync(id, correlationId, envelopeJson, retention, token),
 398            IsTransient,
 399            _options.PublishMaxAttempts,
 400            _options.PublishRetryBaseDelay,
 401            _options.PublishRetryMaxDelay,
 402            cancellationToken);
 403
 404    private async Task<PostgreSqlChannelMessage> InsertMessageOnceAsync(Guid id, string correlationId, string envelopeJs
 405    {
 406        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 407        if (ShouldPrune(ref _lastMessagePruneTicks))
 408            await PruneExpiredMessagesAsync(cancellationToken).ConfigureAwait(false);
 409
 410        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 411        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.
 415        command.CommandText =
 416            $"""
 417            WITH inserted AS (
 418                INSERT INTO {MessageTable} (id, correlation_id, envelope_json, expires_at)
 419                VALUES (@id, @correlation_id, @envelope_json, now() + @retention)
 420                ON CONFLICT (id) DO NOTHING
 421                RETURNING created_at
 422            )
 423            SELECT (SELECT created_at FROM inserted) AS created_at,
 424                   pg_notify(@channel, @payload);
 425            """;
 426        command.Parameters.AddWithValue("id", id);
 427        command.Parameters.AddWithValue("correlation_id", correlationId);
 428        command.Parameters.Add("envelope_json", NpgsqlDbType.Text).Value = envelopeJson;
 429        command.Parameters.AddWithValue("retention", retention);
 430        command.Parameters.AddWithValue("channel", NotificationChannel);
 431        command.Parameters.AddWithValue("payload", NotifyPayload(correlationId));
 432        DateTimeOffset? createdAt;
 433        await using (var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false))
 434        {
 435            await reader.ReadAsync(cancellationToken).ConfigureAwait(false);
 436            createdAt = reader.IsDBNull(0) ? null : reader.GetFieldValue<DateTimeOffset>(0);
 437        }
 438
 439        if (createdAt is { } stamped)
 440            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.
 447        await using var lookup = connection.CreateCommand();
 448        lookup.CommandText = $"SELECT created_at, acked_at, acked_seq FROM {MessageTable} WHERE id = @id;";
 449        lookup.Parameters.AddWithValue("id", id);
 450        await using var existing = await lookup.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 451        if (await existing.ReadAsync(cancellationToken).ConfigureAwait(false))
 452        {
 453            return new PostgreSqlChannelMessage(
 454                id,
 455                correlationId,
 456                envelopeJson,
 457                existing.GetFieldValue<DateTimeOffset>(0),
 458                existing.IsDBNull(1) ? null : existing.GetFieldValue<DateTimeOffset>(1),
 459                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.
 466        throw new InvalidOperationException(
 467            $"PostgreSQL response insert for message {id} found no row after a duplicate: the original no longer exists 
 468    }
 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    {
 478        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 479        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 480        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.
 487        command.CommandText =
 488            $"""
 489            SELECT id, correlation_id, CASE WHEN acked_at IS NULL THEN envelope_json::text END, created_at, acked_at, ac
 490            FROM {MessageTable}
 491            WHERE correlation_id = @correlation_id
 492              AND created_at >= @since
 493              AND expires_at > now()
 494              {(afterCreatedAtUtc is null ? "" : "AND (created_at > @after_created_at OR (created_at = @after_created_at
 495            ORDER BY created_at, id
 496            LIMIT @limit;
 497            """;
 498        command.Parameters.AddWithValue("correlation_id", correlationId);
 499        command.Parameters.AddWithValue("since", sinceUtc);
 500        command.Parameters.AddWithValue("limit", batchSize);
 501        if (afterCreatedAtUtc is not null)
 502        {
 503            command.Parameters.AddWithValue("after_created_at", afterCreatedAtUtc.Value);
 504            command.Parameters.AddWithValue("after_id", afterId ?? throw new ArgumentNullException(nameof(afterId)));
 505        }
 506
 507        return await ReadMessagesAsync(command, batchSize, cancellationToken).ConfigureAwait(false);
 508    }
 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    {
 521        if (ids.Count == 0)
 522            return [];
 523
 524        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 525        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 526        await using var command = connection.CreateCommand();
 527        command.CommandText =
 528            $"""
 529            SELECT id, correlation_id, envelope_json::text, created_at, acked_at, acked_seq
 530            FROM {MessageTable}
 531            WHERE correlation_id = @correlation_id
 532              AND id = ANY(@ids)
 533              AND expires_at > now()
 534            ORDER BY created_at, id;
 535            """;
 536        command.Parameters.AddWithValue("correlation_id", correlationId);
 537        command.Parameters.AddWithValue("ids", ids is Guid[] array ? array : [.. ids]);
 538        return await ReadMessagesAsync(command, ids.Count, cancellationToken).ConfigureAwait(false);
 539    }
 540
 541    private static async Task<IReadOnlyList<PostgreSqlChannelMessage>> ReadMessagesAsync(NpgsqlCommand command, int capa
 542    {
 543        var messages = new List<PostgreSqlChannelMessage>(capacity);
 544        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 545        while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 546            messages.Add(new PostgreSqlChannelMessage(
 547                reader.GetGuid(0),
 548                reader.GetString(1),
 549                reader.IsDBNull(2) ? null : reader.GetString(2),
 550                reader.GetFieldValue<DateTimeOffset>(3),
 551                reader.IsDBNull(4) ? null : reader.GetFieldValue<DateTimeOffset>(4),
 552                reader.IsDBNull(5) ? null : reader.GetInt64(5)));
 553        return messages;
 554    }
 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    {
 565        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 566        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 567        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.
 574        command.CommandText =
 575            $"""
 576            UPDATE {MessageTable}
 577            SET acked_at = COALESCE(acked_at, now()),
 578                acked_seq = CASE WHEN acked_at IS NULL THEN nextval('{AckSequence}') ELSE acked_seq END
 579            WHERE id = @id AND NOT recovery_claimed AND expires_at > now()
 580            RETURNING id;
 581            """;
 582        command.Parameters.AddWithValue("id", messageId);
 583        var result = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
 584        return result is not null and not DBNull;
 585    }
 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    {
 595        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 596        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 597        await using var command = connection.CreateCommand();
 598        command.CommandText =
 599            $"""
 600            UPDATE {MessageTable}
 601            SET recovery_claimed = true
 602            WHERE id = @id AND acked_at IS NULL
 603            RETURNING id;
 604            """;
 605        command.Parameters.AddWithValue("id", messageId);
 606        var result = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
 607        return result is not null and not DBNull;
 608    }
 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    {
 617        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 618        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 619        await using var command = connection.CreateCommand();
 620        command.CommandText = $"SELECT now(), nextval('{AckSequence}');";
 621        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 622        await reader.ReadAsync(cancellationToken).ConfigureAwait(false);
 623        return (reader.GetFieldValue<DateTimeOffset>(0).ToUniversalTime(), reader.GetInt64(1));
 624    }
 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    {
 629        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 630        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 631        await using var command = connection.CreateCommand();
 632        command.CommandText = "SELECT now();";
 633        var result = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
 634        return result switch
 635        {
 636            DateTimeOffset dto => dto.ToUniversalTime(),
 637            DateTime dt => new DateTimeOffset(DateTime.SpecifyKind(dt, DateTimeKind.Utc), TimeSpan.Zero),
 638            _ => DateTimeOffset.UtcNow
 639        };
 640    }
 641
 642    public async Task<bool> IsMessageAcknowledgedAsync(Guid messageId, CancellationToken cancellationToken)
 643    {
 644        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 645        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 646        await using var command = connection.CreateCommand();
 647        command.CommandText = $"SELECT acked_at IS NOT NULL FROM {MessageTable} WHERE id = @id AND expires_at > now();";
 648        command.Parameters.AddWithValue("id", messageId);
 649        var result = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
 650        return result is bool acknowledged && acknowledged;
 651    }
 652
 653    public async Task UpsertSubscriberAsync(string correlationId, Guid registrationId, string instanceId, TimeSpan ttl, 
 654    {
 655        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 656        if (ShouldPrune(ref _lastSubscriberPruneTicks))
 657            await PruneExpiredSubscribersAsync(correlationId, cancellationToken).ConfigureAwait(false);
 658
 659        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 660        await using var command = connection.CreateCommand();
 661        command.CommandText =
 662            $"""
 663            INSERT INTO {SubscriberTable} (correlation_id, registration_id, instance_id, expires_at)
 664            VALUES (@correlation_id, @registration_id, @instance_id, now() + @ttl)
 665            ON CONFLICT (correlation_id, registration_id)
 666            DO UPDATE SET instance_id = EXCLUDED.instance_id,
 667                          expires_at = EXCLUDED.expires_at;
 668            """;
 669        command.Parameters.AddWithValue("correlation_id", correlationId);
 670        command.Parameters.AddWithValue("registration_id", registrationId);
 671        command.Parameters.AddWithValue("instance_id", instanceId);
 672        command.Parameters.AddWithValue("ttl", ttl);
 673        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 674    }
 675
 676    public async Task HeartbeatSubscribersAsync(
 677        string instanceId,
 678        IReadOnlyCollection<(string CorrelationId, Guid RegistrationId)> registrations,
 679        TimeSpan ttl,
 680        CancellationToken cancellationToken)
 681    {
 682        if (registrations.Count == 0)
 683            return;
 684
 685        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 686        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 687        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".
 692        var correlationIds = new string[registrations.Count];
 693        var registrationIds = new Guid[registrations.Count];
 694        var index = 0;
 695        foreach (var (correlationId, registrationId) in registrations)
 696        {
 697            correlationIds[index] = correlationId;
 698            registrationIds[index] = registrationId;
 699            index++;
 700        }
 701
 702        command.CommandText =
 703            $"""
 704            INSERT INTO {SubscriberTable} (correlation_id, registration_id, instance_id, expires_at)
 705            SELECT correlation_id, registration_id, @instance_id, now() + @ttl
 706            FROM unnest(@correlation_ids, @registration_ids) AS live (correlation_id, registration_id)
 707            ON CONFLICT (correlation_id, registration_id)
 708            DO UPDATE SET instance_id = EXCLUDED.instance_id,
 709                          expires_at = EXCLUDED.expires_at;
 710            """;
 711        command.Parameters.AddWithValue("instance_id", instanceId);
 712        command.Parameters.AddWithValue("correlation_ids", NpgsqlDbType.Array | NpgsqlDbType.Text, correlationIds);
 713        command.Parameters.AddWithValue("registration_ids", NpgsqlDbType.Array | NpgsqlDbType.Uuid, registrationIds);
 714        command.Parameters.AddWithValue("ttl", ttl);
 715        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 716    }
 717
 718    public async Task DeleteSubscriberAsync(string correlationId, Guid registrationId, CancellationToken cancellationTok
 719    {
 720        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 721        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 722        await using var command = connection.CreateCommand();
 723        command.CommandText = $"DELETE FROM {SubscriberTable} WHERE correlation_id = @correlation_id AND registration_id
 724        command.Parameters.AddWithValue("correlation_id", correlationId);
 725        command.Parameters.AddWithValue("registration_id", registrationId);
 726        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 727    }
 728
 729    public async Task<long> CountActiveSubscribersAsync(string correlationId, CancellationToken cancellationToken)
 730    {
 731        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 732        if (ShouldPrune(ref _lastSubscriberPruneTicks))
 733            await PruneExpiredSubscribersAsync(correlationId, cancellationToken).ConfigureAwait(false);
 734
 735        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 736        await using var command = connection.CreateCommand();
 737        command.CommandText =
 738            $"""
 739            SELECT count(*)::bigint
 740            FROM {SubscriberTable}
 741            WHERE correlation_id = @correlation_id AND expires_at > now();
 742            """;
 743        command.Parameters.AddWithValue("correlation_id", correlationId);
 744        var result = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
 745        return result is long count ? count : 0L;
 746    }
 747
 748    public async Task ExecuteListenAsync(Func<string?, Task> onNotification, CancellationToken cancellationToken)
 749    {
 750        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 751        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 752        connection.Notification += (_, args) => _ = onNotification(args.Payload);
 753        await using (var command = connection.CreateCommand())
 754        {
 755            command.CommandText = $"LISTEN {Quote(NotificationChannel)};";
 756            await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 757        }
 758
 759        while (!cancellationToken.IsCancellationRequested)
 760            await connection.WaitAsync(cancellationToken).ConfigureAwait(false);
 761    }
 762
 763    private async Task PruneExpiredRecoveryAsync(string? correlationId, CancellationToken cancellationToken)
 764    {
 765        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 766        await using var command = connection.CreateCommand();
 767        command.CommandText = correlationId is null
 768            ? $"DELETE FROM {RecoveryTable} WHERE expires_at <= now();"
 769            : $"DELETE FROM {RecoveryTable} WHERE correlation_id = @correlation_id AND expires_at <= now();";
 770        if (correlationId is not null)
 771            command.Parameters.AddWithValue("correlation_id", correlationId);
 772        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 773    }
 774
 775    private async Task PruneExpiredMessagesAsync(CancellationToken cancellationToken)
 776    {
 777        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 778        await using var command = connection.CreateCommand();
 779        command.CommandText = $"DELETE FROM {MessageTable} WHERE expires_at <= now();";
 780        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 781    }
 782
 783    private async Task PruneExpiredSubscribersAsync(string? correlationId, CancellationToken cancellationToken)
 784    {
 785        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 786        await using var command = connection.CreateCommand();
 787        command.CommandText = correlationId is null
 788            ? $"DELETE FROM {SubscriberTable} WHERE expires_at <= now();"
 789            : $"DELETE FROM {SubscriberTable} WHERE correlation_id = @correlation_id AND expires_at <= now();";
 790        if (correlationId is not null)
 791            command.Parameters.AddWithValue("correlation_id", correlationId);
 792        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 793    }
 794
 795    public static void ValidateIdentifier(string? value, string name)
 796    {
 797        if (string.IsNullOrWhiteSpace(value))
 798            throw new InvalidOperationException($"{nameof(PostgreSqlAsyncResponseChannelOptions)}.{name} must be configu
 799        if (!IsIdentifier(value))
 800            throw new InvalidOperationException(
 801                $"{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.
 804        if (value.Length > IdentifierCap)
 805            throw new InvalidOperationException(
 806                $"{nameof(PostgreSqlAsyncResponseChannelOptions)}.{name} '{value}' is {value.Length} characters; Postgre
 807    }
 808
 809    private static bool IsIdentifier(string value)
 810    {
 811        if (value.Length == 0 || !(char.IsAsciiLetter(value[0]) || value[0] == '_'))
 812            return false;
 813
 814        foreach (var c in value)
 815        {
 816            if (!(char.IsAsciiLetterOrDigit(c) || c == '_'))
 817                return false;
 818        }
 819
 820        return true;
 821    }
 822
 823    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>
 830    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)
 842        => RelationalNamePlan.DerivedName(table, $"_{suffix}_idx", IdentifierCap);
 843
 844    private static string SequenceName(string table)
 845        => 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    {
 859        (string Role, string Name)[] plan =
 860        [
 861            ($"{nameof(options.RecoveryStateTable)} table", options.RecoveryStateTable),
 862            ($"{nameof(options.MessageTable)} table", options.MessageTable),
 863            ($"{nameof(options.SubscriberTable)} table", options.SubscriberTable),
 864            ("ack sequence (derived from MessageTable)", SequenceName(options.MessageTable)),
 865            ("RecoveryStateTable expiry index", IndexName(options.RecoveryStateTable, "expires")),
 866            ("MessageTable correlation index", IndexName(options.MessageTable, "correlation_created")),
 867            ("MessageTable expiry index", IndexName(options.MessageTable, "expires")),
 868            ("SubscriberTable expiry index", IndexName(options.SubscriberTable, "expires")),
 869        ];
 870        RelationalNamePlan.RequireDistinct(
 871            plan,
 872            nameof(PostgreSqlAsyncResponseChannelOptions),
 873            ". All tables and the index/sequence names derived from them share one namespace and must be distinct " +
 874            "(long names reserve suffix space by truncating the table stem, which can make distinct tables derive " +
 875            "the same name). Shorten or de-overlap the configured table names.");
 876    }
 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)
 880        => Encoding.UTF8.GetByteCount(correlationId) <= MaxNotifyPayloadBytes ? correlationId : string.Empty;
 881
 882    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;
 894        var hash = offset;
 895        foreach (var b in Encoding.UTF8.GetBytes($"asyncresponse:ddl:{schemaName}"))
 896        {
 897            hash ^= b;
 898            hash *= prime;
 899        }
 900
 901        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    {
 911        var interval = _options.PruneInterval;
 912        if (interval <= TimeSpan.Zero)
 913            return true;
 914
 915        var now = DateTime.UtcNow.Ticks;
 916        var last = Interlocked.Read(ref lastTicks);
 917        return now - last >= interval.Ticks
 918            && 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

get_Id()
get_CorrelationId()
get_EnvelopeJson()
get_CreatedAtUtc()
get_AckedAtUtc()
get_AckedSeq()