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

Information
Class: AsyncResponse.DurableFlows.PostgreSQL.PostgreSqlFlowStateStore
Assembly: AsyncResponse.DurableFlows.PostgreSQL
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/DurableFlows/AsyncResponse.DurableFlows.PostgreSQL/PostgreSqlDurableFlows.cs
Line coverage
98%
Covered lines: 160
Uncovered lines: 3
Coverable lines: 163
Total lines: 348
Line coverage: 98.1%
Branch coverage
83%
Covered branches: 15
Total branches: 18
Branch coverage: 83.3%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%11100%
LoadAsync()100%11100%
TryCreateAsync()100%22100%
TryUpdateAsync()100%22100%
TryAcquireLeaseAsync(...)100%11100%
TryRenewLeaseAsync(...)100%11100%
ReleaseLeaseAsync()100%11100%
TryDeleteAsync()100%11100%
PruneExpiredAsync()100%11100%
EnsureCreatedAsync()83.33%6697.06%
UpdateLeaseAsync()75%4495%
Dispose()100%22100%
DisposeAsync()50%2275%
get_Schema()100%11100%
get_Table()100%11100%
get_IndexName()100%11100%
Quote(...)100%11100%

File(s)

/home/runner/work/AsyncResponse/AsyncResponse/src/DurableFlows/AsyncResponse.DurableFlows.PostgreSQL/PostgreSqlDurableFlows.cs

#LineLine coverage
 1using AsyncResponse;
 2using AsyncResponse.DurableFlows.Internal;
 3using AsyncResponse.DurableFlows.PostgreSQL;
 4using Microsoft.Extensions.DependencyInjection.Extensions;
 5using Microsoft.Extensions.Options;
 6using Npgsql;
 7using NpgsqlTypes;
 8
 9namespace Microsoft.Extensions.DependencyInjection
 10{
 11    /// <summary>DI registration for the PostgreSQL durable-flow state store.</summary>
 12    public static class PostgreSqlDurableFlowServiceCollectionExtensions
 13    {
 14        /// <summary>
 15        /// Stores durable-flow state in PostgreSQL. Hosts may either register an
 16        /// <see cref="NpgsqlDataSource"/> singleton or set
 17        /// <see cref="PostgreSqlDurableFlowOptions.ConnectionString"/>.
 18        /// </summary>
 19        public static AsyncResponseRegistrationBuilder WithPostgreSqlDurableFlows(
 20            this AsyncResponseRegistrationBuilder builder,
 21            Action<PostgreSqlDurableFlowOptions>? configure = null)
 22        {
 23            // Singleton on purpose: schema provisioning is cached per store instance, and the
 24            // executor resolves the store from a fresh scope per flow execution — a scoped store
 25            // would re-run EnsureCreated's DDL round-trip on every run. All dependencies are
 26            // singletons, so the singleton is safe.
 27            builder.Services.TryAddSingleton(provider =>
 28            {
 29                var options = provider.GetRequiredService<IOptions<PostgreSqlDurableFlowOptions>>();
 30
 31                // Reuse a host-registered NpgsqlDataSource when present; otherwise create one from
 32                // ConnectionString, owned (and disposed) by the store. Nothing is registered as a
 33                // bare NpgsqlDataSource service, so unrelated resolutions of that type are never
 34                // answered — or broken — by this package.
 35                var shared = provider.GetService<NpgsqlDataSource>();
 36                if (shared is not null)
 37                    return new PostgreSqlFlowStateStore(shared, options);
 38
 39                if (string.IsNullOrWhiteSpace(options.Value.ConnectionString))
 40                    throw new InvalidOperationException($"{nameof(PostgreSqlDurableFlowOptions)}.{nameof(PostgreSqlDurab
 41                return new PostgreSqlFlowStateStore(NpgsqlDataSource.Create(options.Value.ConnectionString), options, ow
 42            });
 43            return builder.WithDurableFlows<PostgreSqlFlowStateStore, PostgreSqlDurableFlowOptions>(configure);
 44        }
 45    }
 46}
 47
 48namespace AsyncResponse.DurableFlows.PostgreSQL
 49{
 50/// <summary>Options for the PostgreSQL durable-flow state store.</summary>
 51public sealed class PostgreSqlDurableFlowOptions : DurableFlowOptions
 52{
 53    /// <summary>Optional PostgreSQL connection string used when no <see cref="NpgsqlDataSource"/> is registered.</summa
 54    public string? ConnectionString { get; set; }
 55
 56    /// <summary>Database schema that contains the durable-flow table. Default: <c>public</c>.</summary>
 57    public string SchemaName { get; set; } = "public";
 58
 59    /// <summary>Table storing one durable-flow ledger row per flow id.</summary>
 60    public string TableName { get; set; } = "asyncresponse_flow_state";
 61
 62    /// <summary>Creates the schema, table, and expiry index on first use.</summary>
 63    public bool AutoCreateSchema { get; set; } = true;
 64
 65    /// <summary>
 66    /// How often <see cref="PostgreSqlFlowStateStore.TryCreateAsync"/> opportunistically deletes one
 67    /// bounded batch (1000 rows) of expired rows (loads already treat expired state as absent;
 68    /// pruning bounds table growth). Zero or negative prunes on every save. Default: 5 minutes.
 69    /// </summary>
 70    public TimeSpan PruneInterval { get; set; } = TimeSpan.FromMinutes(5);
 71
 72    /// <summary>
 73    /// Maximum serialized flow-state size in bytes accepted by writes; oversized ledgers fail fast
 74    /// with an actionable error instead of an opaque provider error. Default: <c>null</c>
 75    /// (unlimited — PostgreSQL <c>jsonb</c> is effectively unbounded), settable as an operator budget.
 76    /// </summary>
 77    public long? MaxStateBytes { get; set; }
 78
 79    /// <summary>Validates option values and throws on misconfiguration.</summary>
 80    public void Validate()
 81    {
 82        DurableFlowStoreShared.ValidateIdentifier(SchemaName, $"{nameof(PostgreSqlDurableFlowOptions)}.{nameof(SchemaNam
 83        DurableFlowStoreShared.ValidateIdentifier(TableName, $"{nameof(PostgreSqlDurableFlowOptions)}.{nameof(TableName)
 84        if (MaxStateBytes is <= 0)
 85            throw new InvalidOperationException($"{nameof(PostgreSqlDurableFlowOptions)}.{nameof(MaxStateBytes)} must be
 86    }
 87}
 88
 89/// <summary>PostgreSQL implementation of <see cref="IFlowStateStore"/>.</summary>
 90public sealed class PostgreSqlFlowStateStore : IFlowStateStore, IDisposable, IAsyncDisposable
 91{
 92    private const int PruneBatchSize = 1000;
 93
 94    private readonly NpgsqlDataSource _dataSource;
 95    private readonly PostgreSqlDurableFlowOptions _options;
 396    private readonly SemaphoreSlim _ensureGate = new(1, 1);
 97    private readonly bool _ownsDataSource;
 98    private readonly long _schemaLockKey;
 99    private long _lastPruneTicks;
 100    private bool _created;
 101
 3102    public PostgreSqlFlowStateStore(NpgsqlDataSource dataSource, IOptions<PostgreSqlDurableFlowOptions> options, bool ow
 103    {
 3104        _dataSource = dataSource;
 3105        _options = options.Value;
 3106        _options.Validate();
 3107        _ownsDataSource = ownsDataSource;
 3108        _schemaLockKey = DurableFlowStoreShared.SchemaLockKey(_options.SchemaName);
 3109    }
 110
 111    public async Task<FlowState?> LoadAsync(string flowId, CancellationToken cancellationToken = default)
 112    {
 1113        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 1114        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 115
 116        // All expiry/lease time math in this store runs on the database clock (now()), never an
 117        // app-computed timestamp: with multiple workers, app clock skew beyond the lease window
 118        // would let two nodes both consider a lease expired and double-run a flow.
 1119        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1120        await using var command = connection.CreateCommand();
 1121        command.CommandText = $"SELECT state_json::text, revision FROM {Table} WHERE flow_id = @flow_id AND expires_at_u
 1122        command.Parameters.AddWithValue("flow_id", flowId);
 123
 1124        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 1125        if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 1126            return null;
 127
 1128        return DurableFlowStoreShared.ReadState(flowId, reader.GetString(0), reader.GetInt64(1));
 1129    }
 130
 131    public async Task<bool> TryCreateAsync(string flowId, FlowState state, TimeSpan ttl, CancellationToken cancellationT
 132    {
 1133        DurableFlowStoreShared.ValidateCreate(flowId, state, ttl);
 1134        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "PostgreSQL");
 1135        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 1136        if (DurableFlowStoreShared.ShouldPrune(ref _lastPruneTicks, _options.PruneInterval))
 1137            await PruneExpiredAsync(cancellationToken).ConfigureAwait(false);
 138
 1139        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1140        await using var command = connection.CreateCommand();
 1141        command.CommandText =
 1142            $"""
 1143            INSERT INTO {Table} (flow_id, state_json, expires_at_utc, updated_at_utc, revision)
 1144            VALUES (@flow_id, @state_json, now() + @ttl, now(), @revision)
 1145            ON CONFLICT (flow_id) DO UPDATE
 1146            SET state_json = EXCLUDED.state_json,
 1147                expires_at_utc = EXCLUDED.expires_at_utc,
 1148                updated_at_utc = EXCLUDED.updated_at_utc,
 1149                revision = EXCLUDED.revision,
 1150                lease_id = NULL,
 1151                lease_expires_at_utc = NULL
 1152            WHERE {Table}.expires_at_utc <= now();
 1153            """;
 1154        command.Parameters.AddWithValue("flow_id", flowId);
 1155        command.Parameters.Add("state_json", NpgsqlDbType.Jsonb).Value = stateJson;
 1156        command.Parameters.Add("ttl", NpgsqlDbType.Interval).Value = DurableFlowStoreShared.ServerClockTtl(ttl);
 1157        command.Parameters.AddWithValue("revision", state.Revision);
 1158        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 1159    }
 160
 161    public async Task<bool> TryUpdateAsync(
 162        string flowId,
 163        FlowState state,
 164        long expectedRevision,
 165        TimeSpan ttl,
 166        string? leaseId = null,
 167        CancellationToken cancellationToken = default)
 168    {
 1169        DurableFlowStoreShared.ValidateUpdate(flowId, state, expectedRevision, ttl);
 1170        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "PostgreSQL");
 1171        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 172
 1173        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1174        await using var command = connection.CreateCommand();
 1175        command.CommandText =
 1176            $"""
 1177            UPDATE {Table}
 1178            SET state_json = @state_json,
 1179                expires_at_utc = now() + @ttl,
 1180                updated_at_utc = now(),
 1181                revision = @new_revision
 1182            WHERE flow_id = @flow_id
 1183              AND revision = @expected_revision
 1184              AND expires_at_utc > now()
 1185              AND (@lease_id IS NULL OR (lease_id = @lease_id AND lease_expires_at_utc > now()));
 1186            """;
 1187        command.Parameters.AddWithValue("flow_id", flowId);
 1188        command.Parameters.Add("state_json", NpgsqlDbType.Jsonb).Value = stateJson;
 1189        command.Parameters.Add("ttl", NpgsqlDbType.Interval).Value = DurableFlowStoreShared.ServerClockTtl(ttl);
 1190        command.Parameters.AddWithValue("expected_revision", expectedRevision);
 1191        command.Parameters.AddWithValue("new_revision", state.Revision);
 1192        command.Parameters.AddWithValue("lease_id", NpgsqlDbType.Text, (object?)leaseId ?? DBNull.Value);
 1193        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 1194    }
 195
 196    public Task<bool> TryAcquireLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken canc
 1197        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: true, cancellationToken);
 198
 199    public Task<bool> TryRenewLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken cancel
 1200        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: false, cancellationToken);
 201
 202    public async Task ReleaseLeaseAsync(string flowId, string leaseId, CancellationToken cancellationToken = default)
 203    {
 1204        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 1205        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1206        await using var command = connection.CreateCommand();
 1207        command.CommandText = $"UPDATE {Table} SET lease_id = NULL, lease_expires_at_utc = NULL WHERE flow_id = @flow_id
 1208        command.Parameters.AddWithValue("flow_id", flowId);
 1209        command.Parameters.AddWithValue("lease_id", leaseId);
 1210        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 1211    }
 212
 213    public async Task<bool> TryDeleteAsync(string flowId, CancellationToken cancellationToken = default)
 214    {
 1215        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 1216        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 217
 1218        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1219        await using var command = connection.CreateCommand();
 1220        command.CommandText = $"DELETE FROM {Table} WHERE flow_id = @flow_id;";
 1221        command.Parameters.AddWithValue("flow_id", flowId);
 1222        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 1223    }
 224
 225    private async Task PruneExpiredAsync(CancellationToken cancellationToken)
 226    {
 227        // One bounded batch per prune interval (policy shared by all relational stores): an
 228        // unbatched DELETE over a large expired backlog holds row locks and bloats one
 229        // transaction for the unlucky create that triggered the prune. Loads already filter on
 230        // expiry, so any backlog beyond the batch just waits for the next interval.
 1231        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1232        await using var command = connection.CreateCommand();
 1233        command.CommandText =
 1234            $"""
 1235            DELETE FROM {Table}
 1236            WHERE ctid IN (SELECT ctid FROM {Table} WHERE expires_at_utc <= now() LIMIT {PruneBatchSize});
 1237            """;
 1238        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 1239    }
 240
 241    private async Task EnsureCreatedAsync(CancellationToken cancellationToken)
 242    {
 1243        if (_created || !_options.AutoCreateSchema)
 1244            return;
 245
 1246        await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 247        try
 248        {
 1249            if (_created)
 0250                return;
 251
 1252            await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1253            await using var transaction = await connection.BeginTransactionAsync(cancellationToken).ConfigureAwait(false
 254
 255            // Serialize schema creation across processes. CREATE ... IF NOT EXISTS is not atomic
 256            // against a concurrent create of the same object: two instances starting together both
 257            // pass the existence check and collide on the system catalog ("duplicate key ...
 258            // pg_type_typname_nsp_index"). The transaction-scoped advisory lock (keyed by schema,
 259            // shared with the channel/transport packages) lets one instance build the schema while
 260            // the rest wait and then find it already present.
 1261            await using (var lockCommand = connection.CreateCommand())
 262            {
 1263                lockCommand.Transaction = transaction;
 1264                lockCommand.CommandText = "SELECT pg_advisory_xact_lock(@lock_key);";
 1265                lockCommand.Parameters.AddWithValue("lock_key", _schemaLockKey);
 1266                await lockCommand.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 267            }
 268
 1269            await using var command = connection.CreateCommand();
 1270            command.Transaction = transaction;
 1271            command.CommandText =
 1272                $"""
 1273                CREATE SCHEMA IF NOT EXISTS {Schema};
 1274                CREATE TABLE IF NOT EXISTS {Table} (
 1275                    flow_id text NOT NULL PRIMARY KEY,
 1276                    state_json jsonb NOT NULL,
 1277                    expires_at_utc timestamptz NOT NULL,
 1278                    updated_at_utc timestamptz NOT NULL,
 1279                    revision bigint NOT NULL DEFAULT 0,
 1280                    lease_id text NULL,
 1281                    lease_expires_at_utc timestamptz NULL
 1282                );
 1283                CREATE INDEX IF NOT EXISTS {IndexName} ON {Table} (expires_at_utc);
 1284                """;
 1285            await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 1286            await transaction.CommitAsync(cancellationToken).ConfigureAwait(false);
 1287            _created = true;
 1288        }
 289        finally
 290        {
 1291            _ensureGate.Release();
 292        }
 1293    }
 294
 295    private async Task<bool> UpdateLeaseAsync(
 296        string flowId,
 297        string leaseId,
 298        TimeSpan leaseDuration,
 299        bool acquire,
 300        CancellationToken cancellationToken)
 301    {
 1302        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 1303        ArgumentException.ThrowIfNullOrWhiteSpace(leaseId);
 1304        if (leaseDuration <= TimeSpan.Zero)
 0305            throw new ArgumentOutOfRangeException(nameof(leaseDuration));
 306
 1307        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 1308        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1309        await using var command = connection.CreateCommand();
 310        // Lease fencing runs entirely on the database clock: acquire steals only leases the
 311        // database considers expired, and renew/extend stays relative to now(), so worker clock
 312        // skew can never make two nodes hold the same lease.
 1313        command.CommandText =
 1314            $"""
 1315            UPDATE {Table}
 1316            SET lease_id = @lease_id, lease_expires_at_utc = now() + @lease_duration
 1317            WHERE flow_id = @flow_id
 1318              AND expires_at_utc > now()
 1319              AND {(acquire ? "(lease_id IS NULL OR lease_expires_at_utc <= now() OR lease_id = @lease_id)" : "lease_id 
 1320            """;
 1321        command.Parameters.AddWithValue("flow_id", flowId);
 1322        command.Parameters.AddWithValue("lease_id", leaseId);
 1323        command.Parameters.Add("lease_duration", NpgsqlDbType.Interval).Value = DurableFlowStoreShared.ServerClockTtl(le
 1324        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 1325    }
 326
 327    /// <summary>Disposes the data source when the store created (and therefore owns) it.</summary>
 328    public void Dispose()
 329    {
 2330        _ensureGate.Dispose();
 2331        if (_ownsDataSource)
 2332            _dataSource.Dispose();
 2333    }
 334
 335    /// <inheritdoc cref="Dispose" />
 336    public async ValueTask DisposeAsync()
 337    {
 2338        _ensureGate.Dispose();
 2339        if (_ownsDataSource)
 0340            await _dataSource.DisposeAsync().ConfigureAwait(false);
 2341    }
 342
 1343    private string Schema => Quote(_options.SchemaName);
 1344    private string Table => $"{Schema}.{Quote(_options.TableName)}";
 1345    private string IndexName => Quote($"{_options.TableName}_expires_idx");
 1346    private static string Quote(string identifier) => "\"" + identifier.Replace("\"", "\"\"", StringComparison.Ordinal) 
 347}
 348}
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.DurableFlows.PostgreSQL.PostgreSqlDurableFlowOptions>, bool)
LoadAsync()
TryCreateAsync()
TryUpdateAsync()
TryAcquireLeaseAsync(string, string, System.TimeSpan, System.Threading.CancellationToken)
TryRenewLeaseAsync(string, string, System.TimeSpan, System.Threading.CancellationToken)
ReleaseLeaseAsync()
TryDeleteAsync()
PruneExpiredAsync()
EnsureCreatedAsync()
UpdateLeaseAsync()
Dispose()
DisposeAsync()
get_Schema()
get_Table()
get_IndexName()
Quote(string)