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

Information
Class: AsyncResponse.DurableFlows.PostgreSQL.PostgreSqlDurableFlowOptions
Assembly: AsyncResponse.DurableFlows.PostgreSQL
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/DurableFlows/AsyncResponse.DurableFlows.PostgreSQL/PostgreSqlDurableFlows.cs
Line coverage
100%
Covered lines: 9
Uncovered lines: 0
Coverable lines: 9
Total lines: 348
Line coverage: 100%
Branch coverage
100%
Covered branches: 4
Total branches: 4
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%
Validate()100%44100%

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>
 357    public string SchemaName { get; set; } = "public";
 58
 59    /// <summary>Table storing one durable-flow ledger row per flow id.</summary>
 360    public string TableName { get; set; } = "asyncresponse_flow_state";
 61
 62    /// <summary>Creates the schema, table, and expiry index on first use.</summary>
 363    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>
 370    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    {
 382        DurableFlowStoreShared.ValidateIdentifier(SchemaName, $"{nameof(PostgreSqlDurableFlowOptions)}.{nameof(SchemaNam
 383        DurableFlowStoreShared.ValidateIdentifier(TableName, $"{nameof(PostgreSqlDurableFlowOptions)}.{nameof(TableName)
 384        if (MaxStateBytes is <= 0)
 285            throw new InvalidOperationException($"{nameof(PostgreSqlDurableFlowOptions)}.{nameof(MaxStateBytes)} must be
 386    }
 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;
 96    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
 102    public PostgreSqlFlowStateStore(NpgsqlDataSource dataSource, IOptions<PostgreSqlDurableFlowOptions> options, bool ow
 103    {
 104        _dataSource = dataSource;
 105        _options = options.Value;
 106        _options.Validate();
 107        _ownsDataSource = ownsDataSource;
 108        _schemaLockKey = DurableFlowStoreShared.SchemaLockKey(_options.SchemaName);
 109    }
 110
 111    public async Task<FlowState?> LoadAsync(string flowId, CancellationToken cancellationToken = default)
 112    {
 113        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 114        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.
 119        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 120        await using var command = connection.CreateCommand();
 121        command.CommandText = $"SELECT state_json::text, revision FROM {Table} WHERE flow_id = @flow_id AND expires_at_u
 122        command.Parameters.AddWithValue("flow_id", flowId);
 123
 124        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 125        if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 126            return null;
 127
 128        return DurableFlowStoreShared.ReadState(flowId, reader.GetString(0), reader.GetInt64(1));
 129    }
 130
 131    public async Task<bool> TryCreateAsync(string flowId, FlowState state, TimeSpan ttl, CancellationToken cancellationT
 132    {
 133        DurableFlowStoreShared.ValidateCreate(flowId, state, ttl);
 134        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "PostgreSQL");
 135        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 136        if (DurableFlowStoreShared.ShouldPrune(ref _lastPruneTicks, _options.PruneInterval))
 137            await PruneExpiredAsync(cancellationToken).ConfigureAwait(false);
 138
 139        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 140        await using var command = connection.CreateCommand();
 141        command.CommandText =
 142            $"""
 143            INSERT INTO {Table} (flow_id, state_json, expires_at_utc, updated_at_utc, revision)
 144            VALUES (@flow_id, @state_json, now() + @ttl, now(), @revision)
 145            ON CONFLICT (flow_id) DO UPDATE
 146            SET state_json = EXCLUDED.state_json,
 147                expires_at_utc = EXCLUDED.expires_at_utc,
 148                updated_at_utc = EXCLUDED.updated_at_utc,
 149                revision = EXCLUDED.revision,
 150                lease_id = NULL,
 151                lease_expires_at_utc = NULL
 152            WHERE {Table}.expires_at_utc <= now();
 153            """;
 154        command.Parameters.AddWithValue("flow_id", flowId);
 155        command.Parameters.Add("state_json", NpgsqlDbType.Jsonb).Value = stateJson;
 156        command.Parameters.Add("ttl", NpgsqlDbType.Interval).Value = DurableFlowStoreShared.ServerClockTtl(ttl);
 157        command.Parameters.AddWithValue("revision", state.Revision);
 158        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 159    }
 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    {
 169        DurableFlowStoreShared.ValidateUpdate(flowId, state, expectedRevision, ttl);
 170        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "PostgreSQL");
 171        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 172
 173        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 174        await using var command = connection.CreateCommand();
 175        command.CommandText =
 176            $"""
 177            UPDATE {Table}
 178            SET state_json = @state_json,
 179                expires_at_utc = now() + @ttl,
 180                updated_at_utc = now(),
 181                revision = @new_revision
 182            WHERE flow_id = @flow_id
 183              AND revision = @expected_revision
 184              AND expires_at_utc > now()
 185              AND (@lease_id IS NULL OR (lease_id = @lease_id AND lease_expires_at_utc > now()));
 186            """;
 187        command.Parameters.AddWithValue("flow_id", flowId);
 188        command.Parameters.Add("state_json", NpgsqlDbType.Jsonb).Value = stateJson;
 189        command.Parameters.Add("ttl", NpgsqlDbType.Interval).Value = DurableFlowStoreShared.ServerClockTtl(ttl);
 190        command.Parameters.AddWithValue("expected_revision", expectedRevision);
 191        command.Parameters.AddWithValue("new_revision", state.Revision);
 192        command.Parameters.AddWithValue("lease_id", NpgsqlDbType.Text, (object?)leaseId ?? DBNull.Value);
 193        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 194    }
 195
 196    public Task<bool> TryAcquireLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken canc
 197        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: true, cancellationToken);
 198
 199    public Task<bool> TryRenewLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken cancel
 200        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: false, cancellationToken);
 201
 202    public async Task ReleaseLeaseAsync(string flowId, string leaseId, CancellationToken cancellationToken = default)
 203    {
 204        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 205        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 206        await using var command = connection.CreateCommand();
 207        command.CommandText = $"UPDATE {Table} SET lease_id = NULL, lease_expires_at_utc = NULL WHERE flow_id = @flow_id
 208        command.Parameters.AddWithValue("flow_id", flowId);
 209        command.Parameters.AddWithValue("lease_id", leaseId);
 210        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 211    }
 212
 213    public async Task<bool> TryDeleteAsync(string flowId, CancellationToken cancellationToken = default)
 214    {
 215        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 216        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 217
 218        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 219        await using var command = connection.CreateCommand();
 220        command.CommandText = $"DELETE FROM {Table} WHERE flow_id = @flow_id;";
 221        command.Parameters.AddWithValue("flow_id", flowId);
 222        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 223    }
 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.
 231        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 232        await using var command = connection.CreateCommand();
 233        command.CommandText =
 234            $"""
 235            DELETE FROM {Table}
 236            WHERE ctid IN (SELECT ctid FROM {Table} WHERE expires_at_utc <= now() LIMIT {PruneBatchSize});
 237            """;
 238        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 239    }
 240
 241    private async Task EnsureCreatedAsync(CancellationToken cancellationToken)
 242    {
 243        if (_created || !_options.AutoCreateSchema)
 244            return;
 245
 246        await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 247        try
 248        {
 249            if (_created)
 250                return;
 251
 252            await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 253            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.
 261            await using (var lockCommand = connection.CreateCommand())
 262            {
 263                lockCommand.Transaction = transaction;
 264                lockCommand.CommandText = "SELECT pg_advisory_xact_lock(@lock_key);";
 265                lockCommand.Parameters.AddWithValue("lock_key", _schemaLockKey);
 266                await lockCommand.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 267            }
 268
 269            await using var command = connection.CreateCommand();
 270            command.Transaction = transaction;
 271            command.CommandText =
 272                $"""
 273                CREATE SCHEMA IF NOT EXISTS {Schema};
 274                CREATE TABLE IF NOT EXISTS {Table} (
 275                    flow_id text NOT NULL PRIMARY KEY,
 276                    state_json jsonb NOT NULL,
 277                    expires_at_utc timestamptz NOT NULL,
 278                    updated_at_utc timestamptz NOT NULL,
 279                    revision bigint NOT NULL DEFAULT 0,
 280                    lease_id text NULL,
 281                    lease_expires_at_utc timestamptz NULL
 282                );
 283                CREATE INDEX IF NOT EXISTS {IndexName} ON {Table} (expires_at_utc);
 284                """;
 285            await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 286            await transaction.CommitAsync(cancellationToken).ConfigureAwait(false);
 287            _created = true;
 288        }
 289        finally
 290        {
 291            _ensureGate.Release();
 292        }
 293    }
 294
 295    private async Task<bool> UpdateLeaseAsync(
 296        string flowId,
 297        string leaseId,
 298        TimeSpan leaseDuration,
 299        bool acquire,
 300        CancellationToken cancellationToken)
 301    {
 302        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 303        ArgumentException.ThrowIfNullOrWhiteSpace(leaseId);
 304        if (leaseDuration <= TimeSpan.Zero)
 305            throw new ArgumentOutOfRangeException(nameof(leaseDuration));
 306
 307        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 308        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 309        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.
 313        command.CommandText =
 314            $"""
 315            UPDATE {Table}
 316            SET lease_id = @lease_id, lease_expires_at_utc = now() + @lease_duration
 317            WHERE flow_id = @flow_id
 318              AND expires_at_utc > now()
 319              AND {(acquire ? "(lease_id IS NULL OR lease_expires_at_utc <= now() OR lease_id = @lease_id)" : "lease_id 
 320            """;
 321        command.Parameters.AddWithValue("flow_id", flowId);
 322        command.Parameters.AddWithValue("lease_id", leaseId);
 323        command.Parameters.Add("lease_duration", NpgsqlDbType.Interval).Value = DurableFlowStoreShared.ServerClockTtl(le
 324        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 325    }
 326
 327    /// <summary>Disposes the data source when the store created (and therefore owns) it.</summary>
 328    public void Dispose()
 329    {
 330        _ensureGate.Dispose();
 331        if (_ownsDataSource)
 332            _dataSource.Dispose();
 333    }
 334
 335    /// <inheritdoc cref="Dispose" />
 336    public async ValueTask DisposeAsync()
 337    {
 338        _ensureGate.Dispose();
 339        if (_ownsDataSource)
 340            await _dataSource.DisposeAsync().ConfigureAwait(false);
 341    }
 342
 343    private string Schema => Quote(_options.SchemaName);
 344    private string Table => $"{Schema}.{Quote(_options.TableName)}";
 345    private string IndexName => Quote($"{_options.TableName}_expires_idx");
 346    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()
Validate()