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

Information
Class: AsyncResponse.DurableFlows.PostgreSQL.PostgreSqlFlowStateStore
Assembly: AsyncResponse.DurableFlows.PostgreSQL
File(s): /_/src/DurableFlows/AsyncResponse.DurableFlows.PostgreSQL/PostgreSqlDurableFlows.cs
Line coverage
98%
Covered lines: 231
Uncovered lines: 3
Coverable lines: 234
Total lines: 497
Line coverage: 98.7%
Branch coverage
91%
Covered branches: 31
Total branches: 34
Branch coverage: 91.1%
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%44100%
ValidateCreate(...)50%22100%
TryCreateAsync()100%22100%
TryUpdateAsync()100%22100%
TryAcquireLeaseAsync(...)100%11100%
TryRenewLeaseAsync(...)100%11100%
ReleaseLeaseAsync()100%11100%
ObserveLeaseAsync()100%88100%
TryDeleteAsync()100%11100%
PruneExpiredAsync()100%11100%
EnsureCreatedAsync()80%101096%
TableExistsAsync()100%11100%
UpdateLeaseAsync()100%22100%
Dispose()100%22100%
DisposeAsync()50%2275%
get_Schema()100%11100%
get_Table()100%11100%
get_IndexName()100%11100%
Quote(...)100%11100%
SqlLiteral(...)100%11100%

File(s)

/_/src/DurableFlows/AsyncResponse.DurableFlows.PostgreSQL/PostgreSqlDurableFlows.cs

#LineLine coverage
 1using AsyncResponse;
 2using AsyncResponse.DurableFlows.Internal;
 3using AsyncResponse.DurableFlows.PostgreSQL;
 4using Microsoft.Extensions.DependencyInjection;
 5using Microsoft.Extensions.DependencyInjection.Extensions;
 6using Microsoft.Extensions.Logging;
 7using Microsoft.Extensions.Options;
 8using Npgsql;
 9using NpgsqlTypes;
 10
 11namespace Microsoft.Extensions.DependencyInjection
 12{
 13    /// <summary>DI registration for the PostgreSQL durable-flow state store.</summary>
 14    public static class PostgreSqlDurableFlowServiceCollectionExtensions
 15    {
 16        /// <summary>
 17        /// Stores durable-flow state in PostgreSQL. Hosts may either register an
 18        /// <see cref="NpgsqlDataSource"/> singleton or set
 19        /// <see cref="PostgreSqlDurableFlowOptions.ConnectionString"/>.
 20        /// </summary>
 21        public static AsyncResponseRegistrationBuilder WithPostgreSqlDurableFlows(
 22            this AsyncResponseRegistrationBuilder builder,
 23            Action<PostgreSqlDurableFlowOptions>? configure = null)
 24        {
 25            // Singleton on purpose: schema provisioning is cached per store instance, and the
 26            // executor resolves the store from a fresh scope per flow execution — a scoped store
 27            // would re-run EnsureCreated's DDL round-trip on every run. All dependencies are
 28            // singletons, so the singleton is safe.
 29            builder.Services.TryAddSingleton(provider =>
 30            {
 31                var options = provider.GetRequiredService<IOptions<PostgreSqlDurableFlowOptions>>();
 32
 33                // Reuse a host-registered NpgsqlDataSource when present; otherwise create one from
 34                // ConnectionString, owned (and disposed) by the store. Nothing is registered as a
 35                // bare NpgsqlDataSource service, so unrelated resolutions of that type are never
 36                // answered — or broken — by this package.
 37                var shared = provider.GetService<NpgsqlDataSource>();
 38                var logger = provider.GetService<ILogger<PostgreSqlFlowStateStore>>();
 39                if (shared is not null)
 40                    return new PostgreSqlFlowStateStore(shared, options, logger: logger);
 41
 42                if (string.IsNullOrWhiteSpace(options.Value.ConnectionString))
 43                    throw new InvalidOperationException($"{nameof(PostgreSqlDurableFlowOptions)}.{nameof(PostgreSqlDurab
 44                return new PostgreSqlFlowStateStore(NpgsqlDataSource.Create(options.Value.ConnectionString), options, ow
 45            });
 46            return builder.WithDurableFlows<PostgreSqlFlowStateStore, PostgreSqlDurableFlowOptions>(configure);
 47        }
 48    }
 49}
 50
 51namespace AsyncResponse.DurableFlows.PostgreSQL
 52{
 53/// <summary>Options for the PostgreSQL durable-flow state store.</summary>
 54public sealed class PostgreSqlDurableFlowOptions : DurableFlowOptions
 55{
 56    /// <summary>Optional PostgreSQL connection string used when no <see cref="NpgsqlDataSource"/> is registered.</summa
 57    public string? ConnectionString { get; set; }
 58
 59    /// <summary>Database schema that contains the durable-flow table. Default: <c>public</c>.</summary>
 60    public string SchemaName { get; set; } = "public";
 61
 62    /// <summary>Table storing one durable-flow ledger row per flow id.</summary>
 63    public string TableName { get; set; } = "asyncresponse_flow_state";
 64
 65    /// <summary>Creates the schema, table, and expiry index on first use.</summary>
 66    public bool AutoCreateSchema { get; set; } = true;
 67
 68    /// <summary>
 69    /// How often <see cref="PostgreSqlFlowStateStore.TryCreateAsync"/> opportunistically deletes one
 70    /// bounded batch (1000 rows) of expired rows (loads already treat expired state as absent;
 71    /// pruning bounds table growth). Zero or negative prunes on every save. Default: 5 minutes.
 72    /// </summary>
 73    public TimeSpan PruneInterval { get; set; } = TimeSpan.FromMinutes(5);
 74
 75    /// <summary>
 76    /// Wall-clock budget one opportunistic prune may spend draining expired rows in batches of
 77    /// 1000 after its first batch (the first always runs). A single batch per interval capped
 78    /// cleanup at ~3 rows/second, which any busier instance outgrew forever; the prune now drains
 79    /// batches until one comes back short or this budget lapses, and reports the outcome on the
 80    /// <c>AsyncResponse</c> meter (<c>asyncresponse.flow_state.pruned_rows</c>,
 81    /// <c>prune_failures</c>, <c>prune_budget_exhausted</c>) and the store's logger. The create
 82    /// that triggers the prune waits for it, so this bounds that create's added latency. Zero
 83    /// keeps the historical single batch. Default: 2 seconds.
 84    /// </summary>
 85    public TimeSpan PruneBudget { get; set; } = DurableFlowStoreShared.DefaultPruneBudget;
 86
 87    /// <summary>
 88    /// Maximum serialized flow-state size in bytes accepted by writes; oversized ledgers fail fast
 89    /// with an actionable error instead of an opaque provider error. Default: <c>null</c>
 90    /// (unlimited — PostgreSQL <c>jsonb</c> is effectively unbounded), settable as an operator budget.
 91    /// </summary>
 92    public long? MaxStateBytes { get; set; }
 93
 94    /// <summary>Validates option values and throws on misconfiguration.</summary>
 95    public void Validate()
 96    {
 97        DurableFlowStoreShared.ValidateIdentifier(SchemaName, $"{nameof(PostgreSqlDurableFlowOptions)}.{nameof(SchemaNam
 98        DurableFlowStoreShared.ValidateIdentifier(TableName, $"{nameof(PostgreSqlDurableFlowOptions)}.{nameof(TableName)
 99
 100        // Indexes share PostgreSQL's relation namespace with tables: a table whose name ends
 101        // exactly where the reserved "_expires_idx" stem truncates derives its own name, and
 102        // CREATE INDEX IF NOT EXISTS would silently match the table and skip the index.
 103        if (string.Equals(DurableFlowStoreShared.DerivedName(TableName, "_expires_idx", 63), TableName, StringComparison
 104            throw new InvalidOperationException(
 105                $"{nameof(PostgreSqlDurableFlowOptions)}.{nameof(TableName)} '{TableName}' collides with its derived exp
 106        DurableFlowStoreShared.ValidateMaxStateBytes(MaxStateBytes, nameof(PostgreSqlDurableFlowOptions));
 107        DurableFlowStoreShared.ValidatePruneBudget(PruneBudget, nameof(PostgreSqlDurableFlowOptions));
 108    }
 109}
 110
 111/// <summary>PostgreSQL implementation of <see cref="IFlowStateStore"/>.</summary>
 112public sealed class PostgreSqlFlowStateStore : IFlowStateStore, IDisposable, IAsyncDisposable
 113{
 114    private readonly ILogger<PostgreSqlFlowStateStore>? _logger;
 115
 116    private readonly NpgsqlDataSource _dataSource;
 117    private readonly PostgreSqlDurableFlowOptions _options;
 211118    private readonly SemaphoreSlim _ensureGate = new(1, 1);
 119    private readonly bool _ownsDataSource;
 120    private readonly long _schemaLockKey;
 121    private long _lastPruneTicks;
 122    private volatile bool _created;
 123
 211124    public PostgreSqlFlowStateStore(NpgsqlDataSource dataSource, IOptions<PostgreSqlDurableFlowOptions> options, bool ow
 125    {
 211126        _dataSource = dataSource;
 211127        _logger = logger;
 211128        _options = options.Value;
 211129        _options.Validate();
 211130        _ownsDataSource = ownsDataSource;
 211131        _schemaLockKey = DurableFlowStoreShared.SchemaLockKey(_options.SchemaName);
 211132    }
 133
 134    public async Task<FlowState?> LoadAsync(string flowId, CancellationToken cancellationToken = default)
 135    {
 673136        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 673137        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 138
 139        // All expiry/lease time math in this store runs on the database clock (now()), never an
 140        // app-computed timestamp: with multiple workers, app clock skew beyond the lease window
 141        // would let two nodes both consider a lease expired and double-run a flow.
 671142        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 671143        await using var command = connection.CreateCommand();
 671144        command.CommandText = $"SELECT state_json::text, revision FROM {Table} WHERE flow_id = @flow_id AND expires_at_u
 671145        command.Parameters.AddWithValue("flow_id", flowId);
 146
 671147        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 671148        if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 4149            return null;
 150
 667151        return DurableFlowStoreShared.ReadState(flowId, reader.GetString(0), reader.GetInt64(1));
 669152    }
 153
 154    /// <inheritdoc />
 155    public void ValidateCreate(string flowId, FlowState state, TimeSpan ttl)
 156    {
 136157        DurableFlowStoreShared.ValidateCreate(flowId, state, ttl);
 136158        if (_options.MaxStateBytes is not null)
 4159            _ = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "PostgreSQL");
 134160    }
 161
 162    public async Task<bool> TryCreateAsync(string flowId, FlowState state, TimeSpan ttl, CancellationToken cancellationT
 163    {
 291164        DurableFlowStoreShared.ValidateCreate(flowId, state, ttl);
 290165        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "PostgreSQL");
 290166        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 289167        if (DurableFlowStoreShared.ShouldPrune(ref _lastPruneTicks, _options.PruneInterval))
 268168            await DurableFlowStoreShared.PruneQuietlyAsync(() => PruneExpiredAsync(cancellationToken), _options.PruneBud
 169
 289170        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 289171        await using var command = connection.CreateCommand();
 289172        command.CommandText =
 289173            $"""
 289174            INSERT INTO {Table} (flow_id, state_json, expires_at_utc, updated_at_utc, revision)
 289175            VALUES (@flow_id, @state_json, now() + @ttl, now(), @revision)
 289176            ON CONFLICT (flow_id) DO UPDATE
 289177            SET state_json = EXCLUDED.state_json,
 289178                expires_at_utc = EXCLUDED.expires_at_utc,
 289179                updated_at_utc = EXCLUDED.updated_at_utc,
 289180                revision = EXCLUDED.revision,
 289181                lease_id = NULL,
 289182                lease_expires_at_utc = NULL
 289183            WHERE {Table}.expires_at_utc <= now();
 289184            """;
 289185        command.Parameters.AddWithValue("flow_id", flowId);
 289186        command.Parameters.Add("state_json", NpgsqlDbType.Text).Value = stateJson;
 289187        command.Parameters.Add("ttl", NpgsqlDbType.Interval).Value = DurableFlowStoreShared.ServerClockTtl(ttl);
 289188        command.Parameters.AddWithValue("revision", state.Revision);
 289189        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 289190    }
 191
 192    public async Task<bool> TryUpdateAsync(
 193        string flowId,
 194        FlowState state,
 195        long expectedRevision,
 196        TimeSpan ttl,
 197        string? leaseId = null,
 198        CancellationToken cancellationToken = default)
 199    {
 864200        DurableFlowStoreShared.ValidateUpdate(flowId, state, expectedRevision, ttl);
 864201        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "PostgreSQL");
 864202        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 203
 864204        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 864205        await using var command = connection.CreateCommand();
 864206        command.CommandText =
 864207            $"""
 864208            UPDATE {Table}
 864209            SET state_json = @state_json,
 864210                expires_at_utc = now() + @ttl,
 864211                updated_at_utc = now(),
 864212                revision = @new_revision
 864213            WHERE flow_id = @flow_id
 864214              AND revision = @expected_revision
 864215              AND expires_at_utc > now()
 864216              AND (@lease_id IS NULL OR (lease_id = @lease_id AND lease_expires_at_utc > now()));
 864217            """;
 864218        command.Parameters.AddWithValue("flow_id", flowId);
 864219        command.Parameters.Add("state_json", NpgsqlDbType.Text).Value = stateJson;
 864220        command.Parameters.Add("ttl", NpgsqlDbType.Interval).Value = DurableFlowStoreShared.ServerClockTtl(ttl);
 864221        command.Parameters.AddWithValue("expected_revision", expectedRevision);
 864222        command.Parameters.AddWithValue("new_revision", state.Revision);
 864223        command.Parameters.AddWithValue("lease_id", NpgsqlDbType.Text, (object?)leaseId ?? DBNull.Value);
 864224        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 864225    }
 226
 227    public Task<bool> TryAcquireLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken canc
 148228        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: true, cancellationToken);
 229
 230    public Task<bool> TryRenewLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken cancel
 9231        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: false, cancellationToken);
 232
 233    public async Task ReleaseLeaseAsync(string flowId, string leaseId, CancellationToken cancellationToken = default)
 234    {
 142235        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 142236        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 142237        await using var command = connection.CreateCommand();
 142238        command.CommandText = $"UPDATE {Table} SET lease_id = NULL, lease_expires_at_utc = NULL WHERE flow_id = @flow_id
 142239        command.Parameters.AddWithValue("flow_id", flowId);
 142240        command.Parameters.AddWithValue("lease_id", leaseId);
 142241        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 142242    }
 243
 244    /// <inheritdoc />
 245    public async Task<FlowLeaseObservation?> ObserveLeaseAsync(string flowId, CancellationToken cancellationToken = defa
 246    {
 18247        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 12248        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 249
 250        // The two lease columns exactly as stored — deliberately no now() predicate, unlike every
 251        // other statement in this store: an expired lease nobody has taken over must keep reading
 252        // as the same lease, because the engine's proof of a live holder is that two observations
 253        // DIFFER. Whether it has lapsed stays UpdateLeaseAsync's call, on the database clock.
 12254        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 12255        await using var command = connection.CreateCommand();
 12256        command.CommandText = $"SELECT lease_id, lease_expires_at_utc FROM {Table} WHERE flow_id = @flow_id;";
 12257        command.Parameters.AddWithValue("flow_id", flowId);
 258
 12259        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 12260        if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 2261            return FlowLeaseObservation.Unheld;
 262
 10263        return DurableFlowStoreShared.LeaseObservation(
 10264            reader.IsDBNull(0) ? null : reader.GetString(0),
 10265            reader.IsDBNull(1) ? null : reader.GetDateTime(1));
 12266    }
 267
 268    public async Task<bool> TryDeleteAsync(string flowId, CancellationToken cancellationToken = default)
 269    {
 10270        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 10271        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 272
 10273        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 10274        await using var command = connection.CreateCommand();
 10275        command.CommandText = $"DELETE FROM {Table} WHERE flow_id = @flow_id;";
 10276        command.Parameters.AddWithValue("flow_id", flowId);
 10277        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 10278    }
 279
 280    private async Task<int> PruneExpiredAsync(CancellationToken cancellationToken)
 281    {
 282        // One bounded batch per call; DurableFlowStoreShared.PruneQuietlyAsync repeats it under
 283        // the PruneBudget while batches come back full (policy shared by all relational stores): an
 284        // unbatched DELETE over a large expired backlog holds row locks and bloats one
 285        // transaction for the unlucky create that triggered the prune. Loads already filter on
 286        // expiry, so any backlog beyond the batch just waits for the next interval.
 134287        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 134288        await using var command = connection.CreateCommand();
 134289        command.CommandText =
 134290            $"""
 134291            DELETE FROM {Table}
 134292            WHERE ctid IN (SELECT ctid FROM {Table} WHERE expires_at_utc <= now() LIMIT {DurableFlowStoreShared.PruneBat
 134293            """;
 134294        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 134295    }
 296
 297    private async Task EnsureCreatedAsync(CancellationToken cancellationToken)
 298    {
 2148299        if (_created)
 1889300            return;
 301
 259302        await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 303        try
 304        {
 259305            if (_created)
 122306                return;
 307
 137308            await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 137309            await using var transaction = await connection.BeginTransactionAsync(cancellationToken).ConfigureAwait(false
 310
 137311            if (_options.AutoCreateSchema)
 312            {
 313                // Serialize schema creation across processes. CREATE ... IF NOT EXISTS is not atomic
 314                // against a concurrent create of the same object: two instances starting together both
 315                // pass the existence check and collide on the system catalog ("duplicate key ...
 316                // pg_type_typname_nsp_index"). The transaction-scoped advisory lock (keyed by schema,
 317                // shared with the channel/transport packages) lets one instance build the schema while
 318                // the rest wait and then find it already present.
 134319                await using (var lockCommand = connection.CreateCommand())
 320                {
 134321                    lockCommand.Transaction = transaction;
 134322                    lockCommand.CommandText = "SELECT pg_advisory_xact_lock(@lock_key);";
 134323                    lockCommand.Parameters.AddWithValue("lock_key", _schemaLockKey);
 134324                    await lockCommand.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 325                }
 326
 134327                await using var command = connection.CreateCommand();
 134328                command.Transaction = transaction;
 134329                command.CommandText =
 134330                    $"""
 134331                    CREATE SCHEMA IF NOT EXISTS {Schema};
 134332                    CREATE TABLE IF NOT EXISTS {Table} (
 134333                        flow_id text NOT NULL PRIMARY KEY,
 134334                        state_json text NOT NULL,
 134335                        expires_at_utc timestamptz NOT NULL,
 134336                        updated_at_utc timestamptz NOT NULL,
 134337                        revision bigint NOT NULL DEFAULT 0,
 134338                        lease_id text NULL,
 134339                        lease_expires_at_utc timestamptz NULL
 134340                    );
 134341                    CREATE INDEX IF NOT EXISTS {IndexName} ON {Table} (expires_at_utc);
 134342
 134343                    -- jsonb REJECTS the \u0000 escape System.Text.Json emits for U+0000 (SQLSTATE
 134344                    -- 22P05), so a ledger every other store accepts failed every write here: the
 134345                    -- flow could not start, or its checkpoint failed on every attempt until the
 134346                    -- job dead-lettered. Nothing queries INSIDE the ledger (it is read back with
 134347                    -- ::text), so text costs nothing. Guarded so the rewrite happens once.
 134348                    DO $$
 134349                    BEGIN
 134350                        IF EXISTS (
 134351                            SELECT 1 FROM information_schema.columns
 134352                            WHERE table_schema = {SqlLiteral(_options.SchemaName)} AND table_name = {SqlLiteral(_options
 134353                              AND column_name = 'state_json' AND data_type = 'jsonb')
 134354                        THEN
 134355                            ALTER TABLE {Table} ALTER COLUMN state_json TYPE text USING state_json::text;
 134356                        END IF;
 134357                    END $$;
 134358                    """;
 359                try
 360                {
 134361                    await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 134362                }
 0363                catch (PostgresException ex) when (ex.SqlState is PostgresErrorCodes.WrongObjectType or PostgresErrorCod
 364                {
 365                    // E.g. CREATE INDEX ... ON a name that is really another component's index:
 366                    // IF NOT EXISTS skipped the table create, and the dependent statement then hits
 367                    // the wrong relation kind mid-batch — surface the namespace collision instead of
 368                    // the raw "cannot open relation".
 0369                    throw new InvalidOperationException(AsyncResponse.Internal.PostgreSqlRelationVerifier.DdlCollisionMe
 370                }
 134371            }
 3372            else if (!await TableExistsAsync(connection, transaction, cancellationToken).ConfigureAwait(false))
 373            {
 374                // Operator-managed schema and the migration has not run yet: the first query
 375                // surfaces a clear PostgreSQL error (the documented "create it yourself, later"
 376                // workflow), and _created stays unlatched so a later operation re-verifies once
 377                // the migration lands. When the relation DOES exist it flows into the same catalog
 378                // verification the DDL path uses — operator-provisioned schemas are exactly what
 379                // that check exists for.
 380                return;
 381            }
 382
 383            // The flow store can share a schema with the channel and transport stores (and
 384            // unrelated objects), whose derived names its own validation cannot see — and
 385            // IF NOT EXISTS also accepts a same-name index with the WRONG definition, exactly as
 386            // an operator-provisioned table can carry the wrong shape. Verify against the catalog
 387            // that both relations actually ARE what this store reads and writes, definitions
 388            // included (under the shared DDL lock when this build just ran the DDL).
 137389            await AsyncResponse.Internal.PostgreSqlRelationVerifier.VerifyAsync(
 137390                connection,
 137391                transaction,
 137392                _options.SchemaName,
 137393                "durable-flow",
 137394                [
 137395                    new(_options.TableName, 'r', Columns:
 137396                        [
 137397                            new("flow_id", "text", Nullable: false, RequiresDeterministicCollation: true),
 137398                            new("state_json", "text", Nullable: false),
 137399                            new("expires_at_utc", "timestamp with time zone", Nullable: false),
 137400                            new("updated_at_utc", "timestamp with time zone", Nullable: false),
 137401                            new("revision", "bigint", Nullable: false, DefaultExpression: "0"),
 137402                            new("lease_id", "text", Nullable: true),
 137403                            new("lease_expires_at_utc", "timestamp with time zone", Nullable: true),
 137404                        ], PrimaryKey: ["flow_id"]),
 137405                    new(DurableFlowStoreShared.DerivedName(_options.TableName, "_expires_idx", 63), 'i', _options.TableN
 137406                ],
 137407                cancellationToken).ConfigureAwait(false);
 408
 134409            await transaction.CommitAsync(cancellationToken).ConfigureAwait(false);
 134410            _created = true;
 134411        }
 412        finally
 413        {
 259414            _ensureGate.Release();
 415        }
 2145416    }
 417
 418    /// <summary>
 419    /// Reports whether ANY relation occupies the configured name (any relkind: a view or foreign
 420    /// component's object must reach verification, which names the precise wrong-kind reason
 421    /// instead of skipping the checks).
 422    /// </summary>
 423    private async Task<bool> TableExistsAsync(NpgsqlConnection connection, NpgsqlTransaction transaction, CancellationTo
 424    {
 3425        await using var command = connection.CreateCommand();
 3426        command.Transaction = transaction;
 3427        command.CommandText =
 3428            """
 3429            SELECT EXISTS (
 3430                SELECT 1
 3431                FROM pg_catalog.pg_class c
 3432                JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
 3433                WHERE n.nspname = @schema AND c.relname = @table);
 3434            """;
 3435        command.Parameters.AddWithValue("schema", _options.SchemaName);
 3436        command.Parameters.AddWithValue("table", _options.TableName);
 3437        return (bool)(await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false))!;
 3438    }
 439
 440
 441    private async Task<bool> UpdateLeaseAsync(
 442        string flowId,
 443        string leaseId,
 444        TimeSpan leaseDuration,
 445        bool acquire,
 446        CancellationToken cancellationToken)
 447    {
 157448        DurableFlowStoreShared.ValidateLeaseArgs(flowId, leaseId, leaseDuration);
 449
 157450        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 157451        await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 157452        await using var command = connection.CreateCommand();
 453        // Lease fencing runs entirely on the database clock: acquire steals only leases the
 454        // database considers expired, and renew/extend stays relative to now(), so worker clock
 455        // skew can never make two nodes hold the same lease.
 157456        command.CommandText =
 157457            $"""
 157458            UPDATE {Table}
 157459            SET lease_id = @lease_id, lease_expires_at_utc = now() + @lease_duration
 157460            WHERE flow_id = @flow_id
 157461              AND expires_at_utc > now()
 157462              AND {(acquire ? "(lease_id IS NULL OR lease_expires_at_utc <= now() OR lease_id = @lease_id)" : "lease_id 
 157463            """;
 157464        command.Parameters.AddWithValue("flow_id", flowId);
 157465        command.Parameters.AddWithValue("lease_id", leaseId);
 157466        command.Parameters.Add("lease_duration", NpgsqlDbType.Interval).Value = DurableFlowStoreShared.ServerClockTtl(le
 157467        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 157468    }
 469
 470    /// <summary>Disposes the data source when the store created (and therefore owns) it.</summary>
 471    public void Dispose()
 472    {
 6473        _ensureGate.Dispose();
 6474        if (_ownsDataSource)
 4475            _dataSource.Dispose();
 6476    }
 477
 478    /// <inheritdoc cref="Dispose" />
 479    public async ValueTask DisposeAsync()
 480    {
 400481        _ensureGate.Dispose();
 400482        if (_ownsDataSource)
 0483            await _dataSource.DisposeAsync().ConfigureAwait(false);
 400484    }
 485
 3238486    private string Schema => Quote(_options.SchemaName);
 3104487    private string Table => $"{Schema}.{Quote(_options.TableName)}";
 134488    private string IndexName => Quote(DurableFlowStoreShared.DerivedName(_options.TableName, "_expires_idx", 63));
 6476489    private static string Quote(string identifier) => "\"" + identifier.Replace("\"", "\"\"", StringComparison.Ordinal) 
 490
 491    /// <summary>
 492    /// A single-quoted SQL string literal, for the catalog lookups inside the DDL's DO block where
 493    /// a parameter cannot be bound.
 494    /// </summary>
 268495    private static string SqlLiteral(string value) => "'" + value.Replace("'", "''", StringComparison.Ordinal) + "'";
 496}
 497}
Generated by: ReportGenerator 5.5.11.0
09/22/2026 - 09:41:04
GitHub | reportgenerator.io

Methods/Properties

.ctor(Npgsql.NpgsqlDataSource,Microsoft.Extensions.Options.IOptions`1<AsyncResponse.DurableFlows.PostgreSQL.PostgreSqlDurableFlowOptions>,System.Boolean,Microsoft.Extensions.Logging.ILogger`1<AsyncResponse.DurableFlows.PostgreSQL.PostgreSqlFlowStateStore>)
LoadAsync()
ValidateCreate(System.String,AsyncResponse.FlowState,System.TimeSpan)
TryCreateAsync()
TryUpdateAsync()
TryAcquireLeaseAsync(System.String,System.String,System.TimeSpan,System.Threading.CancellationToken)
TryRenewLeaseAsync(System.String,System.String,System.TimeSpan,System.Threading.CancellationToken)
ReleaseLeaseAsync()
ObserveLeaseAsync()
TryDeleteAsync()
PruneExpiredAsync()
EnsureCreatedAsync()
TableExistsAsync()
UpdateLeaseAsync()
Dispose()
DisposeAsync()
get_Schema()
get_Table()
get_IndexName()
Quote(System.String)
SqlLiteral(System.String)