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

Information
Class: AsyncResponse.Channels.PostgreSQL.PostgreSqlChannelSql
Assembly: AsyncResponse.Channels.PostgreSQL
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/Channels/AsyncResponse.Channels.PostgreSQL/PostgreSqlChannelSql.cs
Line coverage
100%
Covered lines: 395
Uncovered lines: 0
Coverable lines: 395
Total lines: 617
Line coverage: 100%
Branch coverage
100%
Covered branches: 88
Total branches: 88
Branch coverage: 100%
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_NotificationChannel()100%11100%
EnsureCreatedAsync()100%66100%
SaveRecoveryStateAsync()100%11100%
LoadRecoveryStatesAsync()100%22100%
DeleteRecoveryStateAsync()100%11100%
ScanRecoveryStateJsonAsync()100%22100%
InsertMessageAsync(...)100%22100%
InsertMessageOnceAsync()100%88100%
LoadMessagesAsync()100%88100%
TryClaimForDeliveryAsync()100%22100%
TryClaimForRecoveryAsync()100%22100%
GetServerTimeUtcAsync()100%22100%
IsMessageAcknowledgedAsync()100%22100%
UpsertSubscriberAsync()100%22100%
HeartbeatSubscribersAsync()100%44100%
DeleteSubscriberAsync()100%11100%
CountActiveSubscribersAsync()100%44100%
ExecuteListenAsync()100%22100%
PruneExpiredRecoveryAsync()100%44100%
PruneExpiredMessagesAsync()100%11100%
PruneExpiredSubscribersAsync()100%44100%
ValidateIdentifier(...)100%44100%
IsIdentifier(...)100%1212100%
Quote(...)100%11100%
IndexName(...)100%22100%
NotifyPayload(...)100%22100%
IsTransient(...)100%66100%
SchemaAdvisoryLockKey(...)100%22100%
ShouldPrune(...)100%44100%

File(s)

/home/runner/work/AsyncResponse/AsyncResponse/src/Channels/AsyncResponse.Channels.PostgreSQL/PostgreSqlChannelSql.cs

#LineLine coverage
 1using Npgsql;
 2using NpgsqlTypes;
 3using System.Text;
 4
 5namespace AsyncResponse.Channels.PostgreSQL;
 6
 7internal readonly record struct PostgreSqlChannelMessage(
 8    Guid Id,
 9    string CorrelationId,
 10    string EnvelopeJson,
 11    DateTimeOffset CreatedAtUtc,
 12    DateTimeOffset? AckedAtUtc = null);
 13
 14/// <summary>SQL helper for the PostgreSQL channel tables and notification channel.</summary>
 15internal sealed class PostgreSqlChannelSql
 16{
 17    // PostgreSQL rejects a NOTIFY payload of 8000 bytes or more; stay well under it. A correlation
 18    // id longer than this is sent as an empty payload, which the listener treats as "scan all".
 19    private const int MaxNotifyPayloadBytes = 7000;
 20
 21    private readonly NpgsqlDataSource _dataSource;
 22    private readonly PostgreSqlAsyncResponseChannelOptions _options;
 323    private readonly SemaphoreSlim _ensureGate = new(1, 1);
 24    private bool _created;
 25    private readonly long _schemaLockKey;
 26    private long _lastRecoveryPruneTicks;
 27    private long _lastMessagePruneTicks;
 28    private long _lastSubscriberPruneTicks;
 29
 330    public PostgreSqlChannelSql(NpgsqlDataSource dataSource, Microsoft.Extensions.Options.IOptions<PostgreSqlAsyncRespon
 31    {
 332        _dataSource = dataSource;
 333        _options = options.Value;
 334        _options.Validate();
 35
 336        Schema = Quote(_options.SchemaName);
 337        RecoveryTable = $"{Schema}.{Quote(_options.RecoveryStateTable)}";
 338        MessageTable = $"{Schema}.{Quote(_options.MessageTable)}";
 339        SubscriberTable = $"{Schema}.{Quote(_options.SubscriberTable)}";
 340        _schemaLockKey = SchemaAdvisoryLockKey(_options.SchemaName);
 341    }
 42
 43    public string Schema { get; }
 44    public string RecoveryTable { get; }
 45    public string MessageTable { get; }
 46    public string SubscriberTable { get; }
 147    public string NotificationChannel => _options.NotificationChannel;
 48
 49    public async Task EnsureCreatedAsync(CancellationToken cancellationToken = default)
 50    {
 351        if (_created || !_options.AutoCreateSchema)
 352            return;
 53
 354        await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 55        try
 56        {
 357            if (_created)
 158                return;
 59
 360            await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 161            await using var transaction = await connection.BeginTransactionAsync(cancellationToken).ConfigureAwait(false
 62
 63            // Serialize schema creation across processes. CREATE ... IF NOT EXISTS is not atomic against a
 64            // concurrent create of the same object: two instances starting together both pass the existence
 65            // check and collide on the system catalog ("duplicate key ... pg_type_typname_nsp_index"). A
 66            // transaction-scoped advisory lock (keyed by schema, shared with the transport store) lets one
 67            // instance build the schema while the rest wait and then find it already present.
 168            await using (var lockCommand = connection.CreateCommand())
 69            {
 170                lockCommand.Transaction = transaction;
 171                lockCommand.CommandText = "SELECT pg_advisory_xact_lock(@lock_key);";
 172                lockCommand.Parameters.AddWithValue("lock_key", _schemaLockKey);
 173                await lockCommand.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 74            }
 75
 176            await using var command = connection.CreateCommand();
 177            command.Transaction = transaction;
 178            command.CommandText =
 179                $"""
 180                CREATE SCHEMA IF NOT EXISTS {Schema};
 181
 182                CREATE TABLE IF NOT EXISTS {RecoveryTable} (
 183                    correlation_id text NOT NULL,
 184                    registration_id uuid NOT NULL,
 185                    state_json jsonb NOT NULL,
 186                    expires_at timestamptz NOT NULL,
 187                    registered_at timestamptz NOT NULL DEFAULT now(),
 188                    PRIMARY KEY (correlation_id, registration_id)
 189                );
 190                CREATE INDEX IF NOT EXISTS {Quote(IndexName(_options.RecoveryStateTable, "expires"))}
 191                    ON {RecoveryTable} (expires_at);
 192
 193                CREATE TABLE IF NOT EXISTS {MessageTable} (
 194                    id uuid PRIMARY KEY,
 195                    correlation_id text NOT NULL,
 196                    envelope_json jsonb NOT NULL,
 197                    created_at timestamptz NOT NULL DEFAULT now(),
 198                    expires_at timestamptz NOT NULL,
 199                    acked_at timestamptz NULL,
 1100                    recovery_claimed boolean NOT NULL DEFAULT false
 1101                );
 1102                ALTER TABLE {MessageTable} ADD COLUMN IF NOT EXISTS recovery_claimed boolean NOT NULL DEFAULT false;
 1103                CREATE INDEX IF NOT EXISTS {Quote(IndexName(_options.MessageTable, "correlation_created"))}
 1104                    ON {MessageTable} (correlation_id, created_at);
 1105                CREATE INDEX IF NOT EXISTS {Quote(IndexName(_options.MessageTable, "expires"))}
 1106                    ON {MessageTable} (expires_at);
 1107
 1108                CREATE TABLE IF NOT EXISTS {SubscriberTable} (
 1109                    correlation_id text NOT NULL,
 1110                    registration_id uuid NOT NULL,
 1111                    instance_id text NOT NULL,
 1112                    expires_at timestamptz NOT NULL,
 1113                    PRIMARY KEY (correlation_id, registration_id)
 1114                );
 1115                CREATE INDEX IF NOT EXISTS {Quote(IndexName(_options.SubscriberTable, "expires"))}
 1116                    ON {SubscriberTable} (expires_at);
 1117                """;
 1118            await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 1119            await transaction.CommitAsync(cancellationToken).ConfigureAwait(false);
 1120            _created = true;
 1121        }
 122        finally
 123        {
 3124            _ensureGate.Release();
 125        }
 3126    }
 127
 128    public async Task SaveRecoveryStateAsync(string correlationId, RecoveryState state, TimeSpan ttl, CancellationToken 
 129    {
 1130        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 1131        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1132        await using var command = connection.CreateCommand();
 1133        command.CommandText =
 1134            $"""
 1135            INSERT INTO {RecoveryTable} (correlation_id, registration_id, state_json, expires_at, registered_at)
 1136            VALUES (@correlation_id, @registration_id, @state_json, now() + @ttl, now())
 1137            ON CONFLICT (correlation_id, registration_id)
 1138            DO UPDATE SET state_json = EXCLUDED.state_json,
 1139                          expires_at = EXCLUDED.expires_at,
 1140                          registered_at = EXCLUDED.registered_at;
 1141            """;
 1142        command.Parameters.AddWithValue("correlation_id", correlationId);
 1143        command.Parameters.AddWithValue("registration_id", state.RegistrationId);
 1144        command.Parameters.Add("state_json", NpgsqlDbType.Jsonb).Value = AsyncResponseJson.Serialize(state);
 1145        command.Parameters.AddWithValue("ttl", ttl);
 1146        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 1147    }
 148
 149    public async Task<IReadOnlyList<string>> LoadRecoveryStatesAsync(string correlationId, CancellationToken cancellatio
 150    {
 1151        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 1152        if (ShouldPrune(ref _lastRecoveryPruneTicks))
 1153            await PruneExpiredRecoveryAsync(correlationId, cancellationToken).ConfigureAwait(false);
 154
 1155        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1156        await using var command = connection.CreateCommand();
 1157        command.CommandText =
 1158            $"""
 1159            SELECT state_json::text
 1160            FROM {RecoveryTable}
 1161            WHERE correlation_id = @correlation_id AND expires_at > now()
 1162            ORDER BY registered_at;
 1163            """;
 1164        command.Parameters.AddWithValue("correlation_id", correlationId);
 165
 1166        var states = new List<string>();
 1167        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 1168        while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 1169            states.Add(reader.GetString(0));
 1170        return states;
 1171    }
 172
 173    public async Task<bool> DeleteRecoveryStateAsync(string correlationId, Guid registrationId, CancellationToken cancel
 174    {
 1175        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 1176        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1177        await using var command = connection.CreateCommand();
 1178        command.CommandText = $"DELETE FROM {RecoveryTable} WHERE correlation_id = @correlation_id AND registration_id =
 1179        command.Parameters.AddWithValue("correlation_id", correlationId);
 1180        command.Parameters.AddWithValue("registration_id", registrationId);
 1181        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 1182    }
 183
 184    public async IAsyncEnumerable<string> ScanRecoveryStateJsonAsync([System.Runtime.CompilerServices.EnumeratorCancella
 185    {
 1186        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 1187        await PruneExpiredRecoveryAsync(null, cancellationToken).ConfigureAwait(false);
 188
 1189        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1190        await using var command = connection.CreateCommand();
 1191        command.CommandText =
 1192            $"""
 1193            SELECT state_json::text
 1194            FROM {RecoveryTable}
 1195            WHERE expires_at > now()
 1196            ORDER BY registered_at;
 1197            """;
 1198        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 1199        while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 1200            yield return reader.GetString(0);
 1201    }
 202
 203    /// <summary>
 204    /// Inserts a response envelope row and notifies listeners. The caller supplies the message id so
 205    /// the insert is idempotent under retry (<c>ON CONFLICT DO NOTHING</c>); the NOTIFY still fires so
 206    /// a retried publish never strands an active waiter. Returns the row's server-stamped
 207    /// <c>created_at</c> (the original row's on a duplicate) so the same-process fast path compares
 208    /// against subscription watermarks on the server clock rather than the app clock.
 209    /// </summary>
 210    public Task<DateTimeOffset> InsertMessageAsync(Guid id, string correlationId, string envelopeJson, TimeSpan retentio
 1211        => AsyncResponseRetry.ExecuteAsync(
 1212            token => InsertMessageOnceAsync(id, correlationId, envelopeJson, retention, token),
 1213            IsTransient,
 1214            _options.PublishMaxAttempts,
 1215            _options.PublishRetryBaseDelay,
 1216            _options.PublishRetryMaxDelay,
 1217            cancellationToken);
 218
 219    private async Task<DateTimeOffset> InsertMessageOnceAsync(Guid id, string correlationId, string envelopeJson, TimeSp
 220    {
 1221        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 1222        if (ShouldPrune(ref _lastMessagePruneTicks))
 1223            await PruneExpiredMessagesAsync(cancellationToken).ConfigureAwait(false);
 224
 1225        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1226        await using var command = connection.CreateCommand();
 227        // Single statement: the final SELECT both fires the NOTIFY exactly once and returns the
 228        // server-stamped created_at — the fresh row's via RETURNING, or the original row's when
 229        // the idempotent insert hit a duplicate.
 1230        command.CommandText =
 1231            $"""
 1232            WITH inserted AS (
 1233                INSERT INTO {MessageTable} (id, correlation_id, envelope_json, expires_at)
 1234                VALUES (@id, @correlation_id, @envelope_json, now() + @retention)
 1235                ON CONFLICT (id) DO NOTHING
 1236                RETURNING created_at
 1237            )
 1238            SELECT COALESCE(
 1239                       (SELECT created_at FROM inserted),
 1240                       (SELECT created_at FROM {MessageTable} WHERE id = @id)) AS created_at,
 1241                   pg_notify(@channel, @payload);
 1242            """;
 1243        command.Parameters.AddWithValue("id", id);
 1244        command.Parameters.AddWithValue("correlation_id", correlationId);
 1245        command.Parameters.Add("envelope_json", NpgsqlDbType.Jsonb).Value = envelopeJson;
 1246        command.Parameters.AddWithValue("retention", retention);
 1247        command.Parameters.AddWithValue("channel", NotificationChannel);
 1248        command.Parameters.AddWithValue("payload", NotifyPayload(correlationId));
 249        DateTimeOffset? createdAt;
 1250        await using (var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false))
 251        {
 1252            await reader.ReadAsync(cancellationToken).ConfigureAwait(false);
 1253            createdAt = reader.IsDBNull(0) ? null : reader.GetFieldValue<DateTimeOffset>(0);
 254        }
 255
 1256        if (createdAt is { } stamped)
 1257            return stamped;
 258
 259        // NULL is (almost always) a CONCURRENT idempotent publish, not a missing row: ON CONFLICT
 260        // detects the other transaction's row against latest data, but the same-statement fallback
 261        // subquery reads under this statement's snapshot, which predates that commit — so the row
 262        // exists and is invisible here (reproduced on PostgreSQL 16). A fresh statement gets a
 263        // fresh read-committed snapshot and resolves it deterministically; no retry loop needed.
 1264        await using var lookup = connection.CreateCommand();
 1265        lookup.CommandText = $"SELECT created_at FROM {MessageTable} WHERE id = @id;";
 1266        lookup.Parameters.AddWithValue("id", id);
 1267        var existing = await lookup.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
 268
 1269        return existing switch
 1270        {
 1271            DateTimeOffset offset => offset,
 1272            DateTime dateTime => new DateTimeOffset(dateTime, TimeSpan.Zero),
 1273
 1274            // Only reachable when the duplicate's original row is genuinely gone (pruned
 1275            // mid-publish): the message is not persisted, and reporting success with a fabricated
 1276            // app-clock timestamp would both lie about persistence and feed a client clock into
 1277            // the server-clock watermark.
 1278            _ => throw new InvalidOperationException(
 1279                $"PostgreSQL response insert for message {id} found no row after a duplicate: the original no longer exi
 1280        };
 1281    }
 282
 283    public async Task<IReadOnlyList<PostgreSqlChannelMessage>> LoadMessagesAsync(
 284        string correlationId,
 285        DateTimeOffset sinceUtc,
 286        int batchSize,
 287        DateTimeOffset? afterCreatedAtUtc,
 288        Guid? afterId,
 289        CancellationToken cancellationToken)
 290    {
 3291        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 3292        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1293        await using var command = connection.CreateCommand();
 1294        command.CommandText =
 1295            $"""
 1296            SELECT id, correlation_id, envelope_json::text, created_at, acked_at
 1297            FROM {MessageTable}
 1298            WHERE correlation_id = @correlation_id
 1299              AND created_at >= @since
 1300              AND expires_at > now()
 1301              {(afterCreatedAtUtc is null ? "" : "AND (created_at > @after_created_at OR (created_at = @after_created_at
 1302            ORDER BY created_at, id
 1303            LIMIT @limit;
 1304            """;
 1305        command.Parameters.AddWithValue("correlation_id", correlationId);
 1306        command.Parameters.AddWithValue("since", sinceUtc);
 1307        command.Parameters.AddWithValue("limit", batchSize);
 1308        if (afterCreatedAtUtc is not null)
 309        {
 1310            command.Parameters.AddWithValue("after_created_at", afterCreatedAtUtc.Value);
 1311            command.Parameters.AddWithValue("after_id", afterId ?? throw new ArgumentNullException(nameof(afterId)));
 312        }
 313
 1314        var messages = new List<PostgreSqlChannelMessage>(batchSize);
 1315        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 1316        while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 1317            messages.Add(new PostgreSqlChannelMessage(
 1318                reader.GetGuid(0),
 1319                reader.GetString(1),
 1320                reader.GetString(2),
 1321                reader.GetFieldValue<DateTimeOffset>(3),
 1322                reader.IsDBNull(4) ? null : reader.GetFieldValue<DateTimeOffset>(4)));
 1323        return messages;
 1324    }
 325
 326    /// <summary>
 327    /// Atomically claims a message for live delivery: sets <c>acked_at</c> unless the publisher has
 328    /// already routed it to the lost-subscriber path (<c>recovery_claimed</c>). Returns <c>false</c>
 329    /// when recovery owns the message, so a slow-but-live waiter does not deliver a response the
 330    /// recovery callback already handled. Multiple processes may each win this claim, preserving
 331    /// cross-process fan-out, because it gates only on <c>recovery_claimed</c>, not on <c>acked_at</c>.
 332    /// </summary>
 333    public async Task<bool> TryClaimForDeliveryAsync(Guid messageId, CancellationToken cancellationToken)
 334    {
 3335        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 3336        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1337        await using var command = connection.CreateCommand();
 1338        command.CommandText =
 1339            $"""
 1340            UPDATE {MessageTable}
 1341            SET acked_at = COALESCE(acked_at, now())
 1342            WHERE id = @id AND NOT recovery_claimed AND expires_at > now()
 1343            RETURNING id;
 1344            """;
 1345        command.Parameters.AddWithValue("id", messageId);
 1346        var result = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
 1347        return result is not null and not DBNull;
 1348    }
 349
 350    /// <summary>
 351    /// Atomically claims a message for the lost-subscriber path: sets <c>recovery_claimed</c> only
 352    /// while no waiter has delivered (<c>acked_at IS NULL</c>). Returns <c>true</c> when recovery wins;
 353    /// <c>false</c> means a live waiter already took the message, so the publisher must not also fire
 354    /// the recovery callback. Row-level locking serializes this against <see cref="TryClaimForDeliveryAsync"/>.
 355    /// </summary>
 356    public async Task<bool> TryClaimForRecoveryAsync(Guid messageId, CancellationToken cancellationToken)
 357    {
 1358        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 1359        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1360        await using var command = connection.CreateCommand();
 1361        command.CommandText =
 1362            $"""
 1363            UPDATE {MessageTable}
 1364            SET recovery_claimed = true
 1365            WHERE id = @id AND acked_at IS NULL
 1366            RETURNING id;
 1367            """;
 1368        command.Parameters.AddWithValue("id", messageId);
 1369        var result = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
 1370        return result is not null and not DBNull;
 1371    }
 372
 373    /// <summary>Returns the database server's current UTC time, used as a clock-safe delivery watermark.</summary>
 374    public async Task<DateTimeOffset> GetServerTimeUtcAsync(CancellationToken cancellationToken)
 375    {
 3376        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 3377        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1378        await using var command = connection.CreateCommand();
 1379        command.CommandText = "SELECT now();";
 1380        var result = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
 1381        return result switch
 1382        {
 1383            DateTimeOffset dto => dto.ToUniversalTime(),
 1384            DateTime dt => new DateTimeOffset(DateTime.SpecifyKind(dt, DateTimeKind.Utc), TimeSpan.Zero),
 1385            _ => DateTimeOffset.UtcNow
 1386        };
 1387    }
 388
 389    public async Task<bool> IsMessageAcknowledgedAsync(Guid messageId, CancellationToken cancellationToken)
 390    {
 3391        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 3392        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1393        await using var command = connection.CreateCommand();
 1394        command.CommandText = $"SELECT acked_at IS NOT NULL FROM {MessageTable} WHERE id = @id AND expires_at > now();";
 1395        command.Parameters.AddWithValue("id", messageId);
 1396        var result = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
 1397        return result is bool acknowledged && acknowledged;
 1398    }
 399
 400    public async Task UpsertSubscriberAsync(string correlationId, Guid registrationId, string instanceId, TimeSpan ttl, 
 401    {
 1402        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 1403        if (ShouldPrune(ref _lastSubscriberPruneTicks))
 1404            await PruneExpiredSubscribersAsync(correlationId, cancellationToken).ConfigureAwait(false);
 405
 1406        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1407        await using var command = connection.CreateCommand();
 1408        command.CommandText =
 1409            $"""
 1410            INSERT INTO {SubscriberTable} (correlation_id, registration_id, instance_id, expires_at)
 1411            VALUES (@correlation_id, @registration_id, @instance_id, now() + @ttl)
 1412            ON CONFLICT (correlation_id, registration_id)
 1413            DO UPDATE SET instance_id = EXCLUDED.instance_id,
 1414                          expires_at = EXCLUDED.expires_at;
 1415            """;
 1416        command.Parameters.AddWithValue("correlation_id", correlationId);
 1417        command.Parameters.AddWithValue("registration_id", registrationId);
 1418        command.Parameters.AddWithValue("instance_id", instanceId);
 1419        command.Parameters.AddWithValue("ttl", ttl);
 1420        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 1421    }
 422
 423    public async Task HeartbeatSubscribersAsync(
 424        string instanceId,
 425        IReadOnlyCollection<(string CorrelationId, Guid RegistrationId)> registrations,
 426        TimeSpan ttl,
 427        CancellationToken cancellationToken)
 428    {
 3429        if (registrations.Count == 0)
 1430            return;
 431
 3432        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 3433        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1434        await using var command = connection.CreateCommand();
 435
 436        // UPSERT rather than a bare UPDATE: the caller only heartbeats registrations that are live
 437        // in this process, so a missing row means the pruner deleted it (e.g. after a >timeout
 438        // stall) — re-creating it here is what brings the waiter back from "permanently invisible".
 1439        var correlationIds = new string[registrations.Count];
 1440        var registrationIds = new Guid[registrations.Count];
 1441        var index = 0;
 1442        foreach (var (correlationId, registrationId) in registrations)
 443        {
 1444            correlationIds[index] = correlationId;
 1445            registrationIds[index] = registrationId;
 1446            index++;
 447        }
 448
 1449        command.CommandText =
 1450            $"""
 1451            INSERT INTO {SubscriberTable} (correlation_id, registration_id, instance_id, expires_at)
 1452            SELECT correlation_id, registration_id, @instance_id, now() + @ttl
 1453            FROM unnest(@correlation_ids, @registration_ids) AS live (correlation_id, registration_id)
 1454            ON CONFLICT (correlation_id, registration_id)
 1455            DO UPDATE SET instance_id = EXCLUDED.instance_id,
 1456                          expires_at = EXCLUDED.expires_at;
 1457            """;
 1458        command.Parameters.AddWithValue("instance_id", instanceId);
 1459        command.Parameters.AddWithValue("correlation_ids", NpgsqlDbType.Array | NpgsqlDbType.Text, correlationIds);
 1460        command.Parameters.AddWithValue("registration_ids", NpgsqlDbType.Array | NpgsqlDbType.Uuid, registrationIds);
 1461        command.Parameters.AddWithValue("ttl", ttl);
 1462        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 1463    }
 464
 465    public async Task DeleteSubscriberAsync(string correlationId, Guid registrationId, CancellationToken cancellationTok
 466    {
 3467        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 3468        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1469        await using var command = connection.CreateCommand();
 1470        command.CommandText = $"DELETE FROM {SubscriberTable} WHERE correlation_id = @correlation_id AND registration_id
 1471        command.Parameters.AddWithValue("correlation_id", correlationId);
 1472        command.Parameters.AddWithValue("registration_id", registrationId);
 1473        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 1474    }
 475
 476    public async Task<long> CountActiveSubscribersAsync(string correlationId, CancellationToken cancellationToken)
 477    {
 3478        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 3479        if (ShouldPrune(ref _lastSubscriberPruneTicks))
 3480            await PruneExpiredSubscribersAsync(correlationId, cancellationToken).ConfigureAwait(false);
 481
 3482        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1483        await using var command = connection.CreateCommand();
 1484        command.CommandText =
 1485            $"""
 1486            SELECT count(*)::bigint
 1487            FROM {SubscriberTable}
 1488            WHERE correlation_id = @correlation_id AND expires_at > now();
 1489            """;
 1490        command.Parameters.AddWithValue("correlation_id", correlationId);
 1491        var result = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
 1492        return result is long count ? count : 0L;
 1493    }
 494
 495    public async Task ExecuteListenAsync(Func<string?, Task> onNotification, CancellationToken cancellationToken)
 496    {
 3497        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 3498        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1499        connection.Notification += (_, args) => _ = onNotification(args.Payload);
 1500        await using (var command = connection.CreateCommand())
 501        {
 1502            command.CommandText = $"LISTEN {Quote(NotificationChannel)};";
 1503            await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 504        }
 505
 1506        while (!cancellationToken.IsCancellationRequested)
 1507            await connection.WaitAsync(cancellationToken).ConfigureAwait(false);
 1508    }
 509
 510    private async Task PruneExpiredRecoveryAsync(string? correlationId, CancellationToken cancellationToken)
 511    {
 1512        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1513        await using var command = connection.CreateCommand();
 1514        command.CommandText = correlationId is null
 1515            ? $"DELETE FROM {RecoveryTable} WHERE expires_at <= now();"
 1516            : $"DELETE FROM {RecoveryTable} WHERE correlation_id = @correlation_id AND expires_at <= now();";
 1517        if (correlationId is not null)
 1518            command.Parameters.AddWithValue("correlation_id", correlationId);
 1519        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 1520    }
 521
 522    private async Task PruneExpiredMessagesAsync(CancellationToken cancellationToken)
 523    {
 1524        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1525        await using var command = connection.CreateCommand();
 1526        command.CommandText = $"DELETE FROM {MessageTable} WHERE expires_at <= now();";
 1527        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 1528    }
 529
 530    private async Task PruneExpiredSubscribersAsync(string? correlationId, CancellationToken cancellationToken)
 531    {
 3532        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1533        await using var command = connection.CreateCommand();
 1534        command.CommandText = correlationId is null
 1535            ? $"DELETE FROM {SubscriberTable} WHERE expires_at <= now();"
 1536            : $"DELETE FROM {SubscriberTable} WHERE correlation_id = @correlation_id AND expires_at <= now();";
 1537        if (correlationId is not null)
 1538            command.Parameters.AddWithValue("correlation_id", correlationId);
 1539        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 1540    }
 541
 542    public static void ValidateIdentifier(string? value, string name)
 543    {
 3544        if (string.IsNullOrWhiteSpace(value))
 3545            throw new InvalidOperationException($"{nameof(PostgreSqlAsyncResponseChannelOptions)}.{name} must be configu
 3546        if (!IsIdentifier(value))
 3547            throw new InvalidOperationException(
 3548                $"{nameof(PostgreSqlAsyncResponseChannelOptions)}.{name} '{value}' must be a simple PostgreSQL identifie
 3549    }
 550
 551    private static bool IsIdentifier(string value)
 552    {
 3553        if (value.Length == 0 || !(char.IsAsciiLetter(value[0]) || value[0] == '_'))
 3554            return false;
 555
 3556        foreach (var c in value)
 557        {
 3558            if (!(char.IsAsciiLetterOrDigit(c) || c == '_'))
 3559                return false;
 560        }
 561
 3562        return true;
 563    }
 564
 3565    private static string Quote(string identifier) => "\"" + identifier + "\"";
 566
 567    private static string IndexName(string table, string suffix)
 568    {
 3569        var name = $"{table}_{suffix}_idx";
 3570        return name.Length <= 63 ? name : name[..63];
 571    }
 572
 573    /// <summary>NOTIFY payload for a publish: the correlation id, or empty when it is too long to carry.</summary>
 574    private static string NotifyPayload(string correlationId)
 3575        => Encoding.UTF8.GetByteCount(correlationId) <= MaxNotifyPayloadBytes ? correlationId : string.Empty;
 576
 577    internal static bool IsTransient(Exception exception)
 3578        => exception is not OperationCanceledException
 3579           && (exception is NpgsqlException { IsTransient: true } || exception is TimeoutException);
 580
 581    /// <summary>
 582    /// Stable 64-bit advisory-lock key for serializing schema creation. Uses FNV-1a over a
 583    /// schema-scoped discriminator: it must be deterministic across processes (so
 584    /// <see cref="string.GetHashCode()"/>, which is per-process randomized, is unusable) and identical
 585    /// to the transport store's key for the same schema so both serialize their shared CREATE SCHEMA.
 586    /// </summary>
 587    internal static long SchemaAdvisoryLockKey(string schemaName)
 588    {
 589        const ulong offset = 14695981039346656037UL;
 590        const ulong prime = 1099511628211UL;
 3591        var hash = offset;
 3592        foreach (var b in Encoding.UTF8.GetBytes($"asyncresponse:ddl:{schemaName}"))
 593        {
 3594            hash ^= b;
 3595            hash *= prime;
 596        }
 597
 3598        return unchecked((long)hash);
 599    }
 600
 601    /// <summary>
 602    /// Time-gates opportunistic pruning so the housekeeping DELETE runs at most once per
 603    /// <see cref="PostgreSqlAsyncResponseChannelOptions.PruneInterval"/> instead of on every operation.
 604    /// Read queries already filter on <c>expires_at</c>, so throttling pruning never affects correctness.
 605    /// </summary>
 606    private bool ShouldPrune(ref long lastTicks)
 607    {
 3608        var interval = _options.PruneInterval;
 3609        if (interval <= TimeSpan.Zero)
 3610            return true;
 611
 3612        var now = DateTime.UtcNow.Ticks;
 3613        var last = Interlocked.Read(ref lastTicks);
 3614        return now - last >= interval.Ticks
 3615            && Interlocked.CompareExchange(ref lastTicks, now, last) == last;
 616    }
 617}
Generated by: ReportGenerator 5.5.10.0
08/03/2026 - 08:59:44
GitHub | reportgenerator.io

Methods/Properties

.ctor(Npgsql.NpgsqlDataSource, Microsoft.Extensions.Options.IOptions<AsyncResponse.Channels.PostgreSQL.PostgreSqlAsyncResponseChannelOptions>)
get_NotificationChannel()
EnsureCreatedAsync()
SaveRecoveryStateAsync()
LoadRecoveryStatesAsync()
DeleteRecoveryStateAsync()
ScanRecoveryStateJsonAsync()
InsertMessageAsync(System.Guid, string, string, System.TimeSpan, System.Threading.CancellationToken)
InsertMessageOnceAsync()
LoadMessagesAsync()
TryClaimForDeliveryAsync()
TryClaimForRecoveryAsync()
GetServerTimeUtcAsync()
IsMessageAcknowledgedAsync()
UpsertSubscriberAsync()
HeartbeatSubscribersAsync()
DeleteSubscriberAsync()
CountActiveSubscribersAsync()
ExecuteListenAsync()
PruneExpiredRecoveryAsync()
PruneExpiredMessagesAsync()
PruneExpiredSubscribersAsync()
ValidateIdentifier(string, string)
IsIdentifier(string)
Quote(string)
IndexName(string, string)
NotifyPayload(string)
IsTransient(System.Exception)
SchemaAdvisoryLockKey(string)
ShouldPrune(ref long)